Abfrage detaillierter Unterschiede zwischen Zeilen für eine große Datenmenge

Ich habe eine Reihe von großen Tabellen mit jeweils> 300 Spalten. Die von mir verwendete Anwendung erstellt "Archive" geänderter Zeilen, indem eine Kopie der aktuellen Zeile in einer sekundären Tabelle erstellt wird.

Betrachten Sie ein einfaches Beispiel:

CREATE TABLE dbo.bigtable
(
  UpdateDate datetime,
  PK varchar(12) PRIMARY KEY,
  col1 varchar(100),
  col2 int,
  col3 varchar(20),
  .
  .
  .
  colN datetime
);

Archivtabelle:

CREATE TABLE dbo.bigtable_archive
(
  UpdateDate datetime,
  PK varchar(12) NOT NULL,
  col1 varchar(100),
  col2 int,
  col3 varchar(20),
  .
  .
  .
  colN datetime
);

Bevor Aktualisierungen ausgeführt werden dbo.bigtable, wird eine Kopie der Zeile in erstellt dbo.bigtable_archiveund anschließend dbo.bigtable.UpdateDatemit dem aktuellen Datum aktualisiert.

Wenn UNIONSie also die beiden Tabellen zusammenfassen und gruppieren, PKwird eine Zeitleiste mit Änderungen erstellt, wenn Sie nach geordnet sind UpdateDate.

Ich möchte einen Bericht erstellen, in dem die Unterschiede zwischen den Zeilen, sortiert nach UpdateDate, gruppiert nach PK, im folgenden Format aufgeführt sind:

PK,   UpdateDate,  ColumnName,  Old Value,   New Value

Old Valueund New Valuekann die relevanten Spalten in ein VARCHAR(MAX)(es sind keine TEXToder BYTESpalten beteiligt) umgewandelt werden, da ich keine Nachbearbeitung der Werte selbst vornehmen muss.

Im Moment kann ich mir keine vernünftige Vorgehensweise für eine große Anzahl von Spalten vorstellen, ohne die Abfragen programmgesteuert zu generieren. Möglicherweise muss ich dies tun.

Offen für viele Ideen, daher werde ich der Frage nach 2 Tagen ein Kopfgeld hinzufügen.

Dies wird nicht besonders hübsch aussehen, angesichts der mehr als 300 Spalten und der Nichtverfügbarkeit von LAG, und es wird wahrscheinlich auch nicht übermäßig gut funktionieren, aber als erstes würde ich den folgenden Ansatz versuchen:

• UNION die zwei Tische.
• Beziehen Sie für jede PK in der kombinierten Menge ihre vorherige "Inkarnation" aus der Archivtabelle (in der folgenden Implementierung wird OUTER APPLY+ TOP (1)als armer Mann verwendet LAG).
• varchar(max)Wandeln Sie jede Datenspalte paarweise in den aktuellen und den vorherigen Wert um und heben Sie die Pivot-Funktion auf ( CROSS APPLY (VALUES ...)funktioniert für diese Operation gut).
• Filtern Sie die Ergebnisse schließlich danach, ob sich die Werte in jedem Paar voneinander unterscheiden.

Das Transact-SQL des oben genannten, wie ich es sehe:

WITH
  Combined AS
  (
    SELECT * FROM dbo.bigtable
    UNION ALL
    SELECT * FROM dbo.bigtable_archive
  ) AS derived,
  OldAndNew AS
  (
    SELECT
      this.*,
      OldCol1 = last.Col1,
      OldCol2 = last.Col2,
      ...
    FROM
      Combined AS this
      OUTER APPLY
      (
        SELECT TOP (1)
          *
        FROM
          dbo.bigtable_archive
        WHERE
          PK = this.PK
          AND UpdateDate < this.UpdateDate
        ORDER BY
          UpdateDate DESC
      ) AS last
  )
SELECT
  t.PK,
  t.UpdateDate,
  x.ColumnName,
  x.OldValue,
  x.NewValue
FROM
  OldAndNew AS t
  CROSS APPLY
  (
    VALUES
    ('Col1', CAST(t.OldCol1 AS varchar(max), CAST(t.Col1 AS varchar(max))),
    ('Col2', CAST(t.OldCol2 AS varchar(max), CAST(t.Col2 AS varchar(max))),
    ...
  ) AS x (ColumnName, OldValue, NewValue)
WHERE
  NOT EXISTS (SELECT x.OldValue INTERSECT x.NewValue)
ORDER BY
  t.PK,
  t.UpdateDate,
  x.ColumnName
;

Wenn Sie die Daten in eine temporäre Tabelle verschieben

create table #T
(
  PK varchar(12) not null,
  UpdateDate datetime not null,
  ColumnName nvarchar(128) not null,
  Value varchar(max),
  Version int not null
);

Sie könnten die Zeilen entsprechen neuen und alten Wert zu finden mit einer selbst auf beitreten PK, ColumnNameund Version = Version + 1.

Der nicht so schöne Teil ist natürlich, dass Sie Ihre 300 Spalten von den beiden Basistabellen in die temporäre Tabelle verschieben.

XML zur Rettung, um die Dinge weniger umständlich zu machen.

Es ist möglich, die Pivotierung von Daten mit XML aufzuheben, ohne zu wissen, welche tatsächlichen Spalten in der Tabelle nicht pivotiert werden. Die Spaltennamen müssen als Elementnamen in XML gültig sein, da sie sonst fehlschlagen.

Die Idee ist, eine XML für jede Zeile zu erstellen, die alle Werte für diese Zeile enthält.

select bt.PK,
       bt.UpdateDate,
       (select bt.* for xml path(''), elements xsinil, type) as X
from dbo.bigtable as bt;

<UpdateDate xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">2001-01-03T00:00:00</UpdateDate>
<PK xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">PK1</PK>
<col1 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">c1_1_3</col1>
<col2 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">3</col2>
<col3 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:nil="true" />
<colN xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">2001-01-03T00:00:00</colN>

elements xsinilgibt es Elemente für Spalten mit zu erstellen NULL.

Das XML kann dann zerkleinert werden, indem nodes('*') für jede Spalte eine Zeile abgerufen und local-name(.)der Elementname abgerufen und text()der Wert abgerufen wird.

  select C1.PK,
         C1.UpdateDate,
         T.X.value('local-name(.)', 'nvarchar(128)') as ColumnName,
         T.X.value('text()[1]', 'varchar(max)') as Value
  from C1
    cross apply C1.X.nodes('row/*') as T(X)

Volle Lösung unten. Beachten Sie, dass dies Versionumgekehrt ist. 0 = Letzte Version.

create table #X
(
  PK varchar(12) not null,
  UpdateDate datetime not null,
  Version int not null,
  RowData xml not null
);

create table #T
(
  PK varchar(12) not null,
  UpdateDate datetime not null,
  ColumnName nvarchar(128) not null,
  Value varchar(max),
  Version int not null
);


insert into #X(PK, UpdateDate, Version, RowData)
select bt.PK,
       bt.UpdateDate,
       0,
       (select bt.* for xml path(''), elements xsinil, type)
from dbo.bigtable as bt
union all
select bt.PK,
       bt.UpdateDate,
       row_number() over(partition by bt.PK order by bt.UpdateDate desc),
       (select bt.* for xml path(''), elements xsinil, type)
from dbo.bigtable_archive as bt;

with C as 
(
  select X.PK,
         X.UpdateDate,
         X.Version,
         T.C.value('local-name(.)', 'nvarchar(128)') as ColumnName,
         T.C.value('text()[1]', 'varchar(max)') as Value
  from #X as X
    cross apply X.RowData.nodes('*') as T(C)
)
insert into #T (PK, UpdateDate, ColumnName, Value, Version)
select C.PK,
       C.UpdateDate,
       C.ColumnName,
       C.Value,
       C.Version
from C 
where C.ColumnName not in (N'PK', N'UpdateDate');

/*
option (querytraceon 8649);

The above query might need some trick to go parallel.
For the testdata I had on my machine exection time is 16 seconds vs 2 seconds
https://sqlkiwi.blogspot.com/2011/12/forcing-a-parallel-query-execution-plan.html
http://dataeducation.com/next-level-parallel-plan-forcing-an-alternative-to-8649/

*/

select New.PK,
       New.UpdateDate,
       New.ColumnName,
       Old.Value as OldValue,
       New.Value as NewValue
from #T as New
  left outer join #T as Old
    on Old.PK = New.PK and
       Old.ColumnName = New.ColumnName and
       Old.Version = New.Version + 1;

Ich würde Ihnen einen anderen Ansatz vorschlagen.

Obwohl Sie die aktuelle Anwendung nicht ändern können, können Sie möglicherweise das Datenbankverhalten ändern.

Wenn möglich, würde ich den aktuellen Tabellen zwei TRIGGERS hinzufügen.

Ein INSTEAD OF INSERT in dbo.bigtable_archive, das den neuen Datensatz nur hinzufügt, wenn er zurzeit nicht vorhanden ist.

CREATE TRIGGER dbo.IoI_BTA
ON dbo.bigtable_archive
INSTEAD OF INSERT
AS
BEGIN
    IF NOT EXISTs(SELECT 1 
                  FROM dbo.bigtable_archive bta
                  INNER JOIN inserted i
                  ON  bta.PK = i.PK
                  AND bta.UpdateDate = i.UpdateDate)
    BEGIN
        INSERT INTO dbo.bigtable_archive
        SELECT * FROM inserted;
    END
END

Und ein AFTER INSERT-Trigger für Bigtable, der genau die gleiche Aufgabe erfüllt, jedoch Daten von Bigtable verwendet.

CREATE TRIGGER dbo.IoI_BT
ON dbo.bigtable
AFTER INSERT
AS
BEGIN
    IF NOT EXISTS(SELECT 1 
                  FROM dbo.bigtable_archive bta
                  INNER JOIN inserted i
                  ON  bta.PK = i.PK
                  AND bta.UpdateDate = i.UpdateDate)
    BEGIN
        INSERT INTO dbo.bigtable_archive
        SELECT * FROM inserted;
    END
END

Ok, ich habe hier ein kleines Beispiel mit diesen Anfangswerten erstellt:

SELECT * FROM bigtable;
SELECT * FROM bigtable_archive;

UpdateDate | PK | col1 | col2 | col3
: ------------------ | : - | : --- | ---: | : ---
02/01/2017 00:00:00 | ABC | C3 | 1 | C1  

UpdateDate | PK | col1 | col2 | col3
: ------------------ | : - | : --- | ---: | : ---
01/01/2017 00:00:00 | ABC | C1 | 1 | C1  

Jetzt sollten Sie in bigtable_archive alle ausstehenden Datensätze von bigtable einfügen.

INSERT INTO bigtable_archive
SELECT *
FROM   bigtable
WHERE  UpdateDate >= '20170102';

SELECT * FROM bigtable_archive;
GO

UpdateDate | PK | col1 | col2 | col3
: ------------------ | : - | : --- | ---: | : ---
01/01/2017 00:00:00 | ABC | C1 | 1 | C1  
02/01/2017 00:00:00 | ABC | C3 | 1 | C1  

Wenn die Anwendung nun das nächste Mal versucht, einen Datensatz in die Tabelle bigtable_archive einzufügen, erkennt der Trigger, ob er vorhanden ist, und das Einfügen wird vermieden.

INSERT INTO dbo.bigtable_archive VALUES('20170102', 'ABC', 'C3', 1, 'C1');
GO

SELECT * FROM bigtable_archive;
GO

UpdateDate | PK | col1 | col2 | col3
: ------------------ | : - | : --- | ---: | : ---
01/01/2017 00:00:00 | ABC | C1 | 1 | C1  
02/01/2017 00:00:00 | ABC | C3 | 1 | C1  

Offensichtlich können Sie jetzt die Zeitleiste der Änderungen abrufen, indem Sie nur die Archivtabelle abfragen. Und die Anwendung wird nie bemerken, dass ein Auslöser leise die Arbeit unter der Decke macht.

dbfiddle hier

Arbeitsvorschlag mit einigen Beispieldaten finden Sie unter @ rextester: bigtable unpivot

Der Kern der Operation:

1 - Verwenden Sie syscolumns und für xml , um unsere Spaltenlisten dynamisch für die Unpivot-Operation zu generieren. Alle Werte werden in varchar (max) konvertiert, wobei NULL-Werte in die Zeichenfolge 'NULL' konvertiert werden.

2 - Generieren Sie eine dynamische Abfrage, um die Pivotierung von Daten in der temporären Tabelle #columns aufzuheben

• Warum eine temporäre Tabelle gegen CTE (via with clause)? befasst sich mit potenziellen Leistungsproblemen bei einem großen Datenvolumen und einem CTE-Self-Join ohne verwendbares Index- / Hashing-Schema; In einer temporären Tabelle kann ein Index erstellt werden, der die Leistung beim Self-Join verbessern soll [siehe langsamer CTE-Self-Join ].
• Daten werden in der Reihenfolge PK + ColName + UpdateDate in #columns geschrieben, sodass wir PK / Colname-Werte in benachbarten Zeilen speichern können. Eine Identitätsspalte ( rid ) ermöglicht es uns, diese aufeinander folgenden Zeilen mit rid = rid + 1 selbst zu verbinden

3 - Führen Sie einen Self-Join der Tabelle #temp durch, um die gewünschte Ausgabe zu generieren

Schneiden-und-Einfügen von rextester ...

Erstellen Sie einige Beispieldaten und unsere Tabelle #columns:

CREATE TABLE dbo.bigtable
(UpdateDate datetime      not null
,PK         varchar(12)   not null
,col1       varchar(100)      null
,col2       int               null
,col3       varchar(20)       null
,col4       datetime          null
,col5       char(20)          null
,PRIMARY KEY (PK)
);

CREATE TABLE dbo.bigtable_archive
(UpdateDate datetime      not null
,PK         varchar(12)   not null
,col1       varchar(100)      null
,col2       int               null
,col3       varchar(20)       null
,col4       datetime          null
,col5       char(20)          null
,PRIMARY KEY (PK, UpdateDate)
);

insert into dbo.bigtable         values ('20170512', 'ABC', NULL, 6, 'C1', '20161223', 'closed')

insert into dbo.bigtable_archive values ('20170427', 'ABC', NULL, 6, 'C1', '20160820', 'open')
insert into dbo.bigtable_archive values ('20170315', 'ABC', NULL, 5, 'C1', '20160820', 'open')
insert into dbo.bigtable_archive values ('20170212', 'ABC', 'C1', 1, 'C1', '20160820', 'open')
insert into dbo.bigtable_archive values ('20170109', 'ABC', 'C1', 1, 'C1', '20160513', 'open')

insert into dbo.bigtable         values ('20170526', 'XYZ', 'sue', 23, 'C1', '20161223', 're-open')

insert into dbo.bigtable_archive values ('20170401', 'XYZ', 'max', 12, 'C1', '20160825', 'cancel')
insert into dbo.bigtable_archive values ('20170307', 'XYZ', 'bob', 12, 'C1', '20160825', 'cancel')
insert into dbo.bigtable_archive values ('20170223', 'XYZ', 'bob', 12, 'C1', '20160820', 'open')
insert into dbo.bigtable_archive values ('20170214', 'XYZ', 'bob', 12, 'C1', '20160513', 'open')
;

create table #columns
(rid        int           identity(1,1)
,PK         varchar(12)   not null
,UpdateDate datetime      not null
,ColName    varchar(128)  not null
,ColValue   varchar(max)      null
,PRIMARY KEY (rid, PK, UpdateDate, ColName)
);

Der Mut der Lösung:

declare @columns_max varchar(max),
        @columns_raw varchar(max),
        @cmd         varchar(max)

select  @columns_max = stuff((select ',isnull(convert(varchar(max),'+name+'),''NULL'') as '+name
                from    syscolumns
                where   id   = object_id('dbo.bigtable')
                and     name not in ('PK','UpdateDate')
                order by name
                for xml path(''))
            ,1,1,''),
        @columns_raw = stuff((select ','+name
                from    syscolumns
                where   id   = object_id('dbo.bigtable')
                and     name not in ('PK','UpdateDate')
                order by name
                for xml path(''))
            ,1,1,'')


select @cmd = '
insert #columns (PK, UpdateDate, ColName, ColValue)
select PK,UpdateDate,ColName,ColValue
from
(select PK,UpdateDate,'+@columns_max+' from bigtable
 union all
 select PK,UpdateDate,'+@columns_max+' from bigtable_archive
) p
unpivot
  (ColValue for ColName in ('+@columns_raw+')
) as unpvt
order by PK, ColName, UpdateDate'

--select @cmd

execute(@cmd)

--select * from #columns order by rid
;

select  c2.PK, c2.UpdateDate, c2.ColName as ColumnName, c1.ColValue as 'Old Value', c2.ColValue as 'New Value'
from    #columns c1,
        #columns c2
where   c2.rid                       = c1.rid + 1
and     c2.PK                        = c1.PK
and     c2.ColName                   = c1.ColName
and     isnull(c2.ColValue,'xxx')   != isnull(c1.ColValue,'xxx')
order by c2.UpdateDate, c2.PK, c2.ColName
;

Und die Ergebnisse:

Hinweis: Entschuldigungen ... konnten keine einfache Möglichkeit zum Ausschneiden und Einfügen der Rextester-Ausgabe in einen Codeblock finden. Ich bin offen für Vorschläge.

Mögliche Probleme / Bedenken:

1 - Die Konvertierung von Daten in ein generisches varchar (max) kann zu einem Verlust der Datengenauigkeit führen, was wiederum bedeuten kann, dass wir einige Datenänderungen verpassen. Berücksichtigen Sie die folgenden datetime- und float-Paare, die beim Konvertieren / Umwandeln in das generische 'varchar (max)' ihre Genauigkeit verlieren (dh die konvertierten Werte sind dieselben):

original value       varchar(max)
-------------------  -------------------
06/10/2017 10:27:15  Jun 10 2017 10:27AM
06/10/2017 10:27:18  Jun 10 2017 10:27AM

    234.23844444                 234.238
    234.23855555                 234.238

    29333488.888            2.93335e+007
    29333499.999            2.93335e+007

Während die Datengenauigkeit aufrechterhalten werden könnte, wäre etwas mehr Codierung erforderlich (z. B. Casting basierend auf Quellenspaltendatentypen). Im Moment habe ich mich dafür entschieden, den generischen varchar (max) gemäß der Empfehlung des OP beizubehalten (und davon auszugehen, dass das OP die Daten gut genug kennt, um zu wissen, dass wir keine Probleme mit Datengenauigkeitsverlusten bekommen).

2 - bei sehr großen Datenmengen besteht die Gefahr, dass einige Serverressourcen aufgebraucht werden, unabhängig davon, ob es sich um temporären Speicherplatz und / oder Cache / Speicher handelt. Das Hauptproblem ergibt sich aus der Datenexplosion, die während eines Unpivots auftritt (z. B. gehen wir von 1 Zeile und 302 Datenelementen zu 300 Zeilen und 1200-1500 Datenelementen über, einschließlich 300 Kopien der PK- und UpdateDate-Spalten und 300 Spaltennamen).

Bei diesem Ansatz wird eine dynamische Abfrage verwendet, um eine SQL-Datei zum Abrufen der Änderungen zu generieren. Der SP verwendet einen Tabellen- und Schemanamen und gibt die gewünschte Ausgabe aus.

Es wird davon ausgegangen, dass die Spalten PK und UpdateDate in allen Tabellen vorhanden sind. Und alle Archivtabellen haben das Format originalTableName + "_archive".

NB: Ich habe es nicht auf Leistung überprüft.

NB: da dies dynamisches SQL verwendet, sollte ich eine Einschränkung in Bezug auf Sicherheit / SQL-Injektion hinzufügen. Beschränken Sie den Zugriff auf SP und fügen Sie weitere Validierungen hinzu, um die SQL-Injektion zu verhindern.

    CREATE proc getTableChanges
    @schemaname  varchar(255),
    @tableName varchar(255)
    as

    declare @strg nvarchar(max), @colNameStrg nvarchar(max)='', @oldValueString nvarchar(max)='', @newValueString nvarchar(max)=''

    set @strg = '
    with cte as (

    SELECT  * , ROW_NUMBER() OVER(partition by PK ORDER BY UpdateDate) as RowNbr
    FROM    (

        SELECT  *
        FROM    [' + @schemaname + '].[' + @tableName + ']

        UNION

        SELECT  *
        FROM    [' + @schemaname + '].[' + @tableName + '_archive]

        ) a

    )
    '


    SET @strg = @strg + '

    SELECT  a.pk, a.updateDate, 
    CASE '

    DECLARE @colName varchar(255)
    DECLARE cur CURSOR FOR
        SELECT  COLUMN_NAME
        FROM    INFORMATION_SCHEMA.COLUMNS
        WHERE TABLE_SCHEMA = @schemaname
        AND TABLE_NAME = @tableName
        AND COLUMN_NAME NOT IN ('PK', 'Updatedate')

    OPEN cur
    FETCH NEXT FROM cur INTO @colName 

    WHILE @@FETCH_STATUS = 0
    BEGIN

        SET @colNameStrg  = @colNameStrg  + ' when a.' + @colName + ' <> b.' + @colName + ' then ''' + @colName + ''' '
        SET @oldValueString = @oldValueString + ' when a.' + @colName + ' <> b.' + @colName + ' then cast(a.' + @colName + ' as varchar(max))'
        SET @newValueString = @newValueString + ' when a.' + @colName + ' <> b.' + @colName + ' then cast(b.' + @colName + ' as varchar(max))'


    FETCH NEXT FROM cur INTO @colName 
    END

    CLOSE cur
    DEALLOCATE cur


    SET @colNameStrg = @colNameStrg  + '    END as ColumnChanges '
    SET @oldValueString = 'CASE ' + @oldValueString + ' END as OldValue'
    SET @newValueString = 'CASE ' + @newValueString + ' END as NewValue'

    SET @strg = @strg + @colNameStrg + ',' + @oldValueString + ',' + @newValueString

    SET @strg = @strg + '
        FROM    cte a join cte b on a.PK = b.PK and a.RowNbr + 1 = b.RowNbr 
        ORDER BY  a.pk, a.UpdateDate
    '

    print @strg

    execute sp_executesql @strg


    go

Beispielanruf:

exec getTableChanges 'dbo', 'bigTable'

In meinem Beispiel verwende ich AdventureWorks2012`, Production.ProductCostHistory und Production.ProductListPriceHistory. Es ist möglicherweise kein perfektes Beispiel für eine Verlaufstabelle, "aber das Skript kann die gewünschte Ausgabe und die korrekte Ausgabe zusammenstellen".

     DECLARE @sql NVARCHAR(MAX)
    ,@columns NVARCHAR(Max)
    ,@table VARCHAR(200) = 'ProductCostHistory'
    ,@Schema VARCHAR(200) = 'Production'
    ,@Archivecolumns NVARCHAR(Max)
    ,@ColForUnpivot NVARCHAR(Max)
    ,@ArchiveColForUnpivot NVARCHAR(Max)
    ,@PKCol VARCHAR(200) = 'ProductID'
    ,@UpdatedCol VARCHAR(200) = 'modifiedDate'
    ,@Histtable VARCHAR(200) = 'ProductListPriceHistory'
SELECT @columns = STUFF((
            SELECT ',CAST(p.' + QUOTENAME(column_name) + ' AS VARCHAR(MAX)) AS ' + QUOTENAME(column_name)
            FROM information_schema.columns
            WHERE table_name = @table
                AND column_name NOT IN (
                    @PKCol
                    ,@UpdatedCol
                    )
            ORDER BY ORDINAL_POSITION
            FOR XML PATH('')
            ), 1, 1, '')
    ,@Archivecolumns = STUFF((
            SELECT ',CAST(p1.' + QUOTENAME(column_name) + ' AS VARCHAR(MAX)) AS ' + QUOTENAME('A_' + column_name)
            FROM information_schema.columns
            WHERE table_name = @Histtable
                AND column_name NOT IN (
                    @PKCol
                    ,@UpdatedCol
                    )
            ORDER BY ORDINAL_POSITION
            FOR XML PATH('')
            ), 1, 1, '')
    ,@ColForUnpivot = STUFF((
            SELECT ',' + QUOTENAME(column_name)
            FROM information_schema.columns
            WHERE table_name = @table
                AND column_name NOT IN (
                    @PKCol
                    ,@UpdatedCol
                    )
            ORDER BY ORDINAL_POSITION
            FOR XML PATH('')
            ), 1, 1, '')
    ,@ArchiveColForUnpivot = STUFF((
            SELECT ',' + QUOTENAME('A_' + column_name)
            FROM information_schema.columns
            WHERE table_name = @Histtable
                AND column_name NOT IN (
                    @PKCol
                    ,@UpdatedCol
                    )
            ORDER BY ORDINAL_POSITION
            FOR XML PATH('')
            ), 1, 1, '')

--SELECT @columns   ,@Archivecolumns    ,@ColForUnpivot
SET @sql = N' 
    SELECT ' + @PKCol + ', ColumnName,
            OldValue,NewValue,' + @UpdatedCol + '
    FROM    (  
    SELECT p.' + @PKCol + '
        ,p.' + @UpdatedCol + '
        ,' + @columns + '
        ,' + @Archivecolumns + '
    FROM ' + @Schema + '.' + @table + ' p
    left JOIN ' + @Schema + '.' + @Histtable + ' p1 ON p.' + @PKCol + ' = p1.' + @PKCol + '

  ) t
    UNPIVOT (
        OldValue
        FOR ColumnName in (' + @ColForUnpivot + ')
    ) up

     UNPIVOT (
        NewValue
        FOR ColumnName1 in (' + @ArchiveColForUnpivot + ')
    ) up1

--print @sql
EXEC (@sql)

Hier in der inneren Auswahlabfrage betrachten Sie p als Haupttabelle und p1 als Verlaufstabelle. Beim Deaktivieren ist es wichtig, es in denselben Typ zu konvertieren.

Sie können einen beliebigen anderen Tabellennamen mit weniger Spaltennamen verwenden, um mein Skript zu verstehen. Für jede Erläuterung muss dann ein Ping an mich gesendet werden.