SQL：为每列的记录返回审计答案

【问题标题】：SQL: Return audit for records for each columnSQL：为每列的记录返回审计
【发布时间】：2012-05-28 07:54:23
【问题描述】：

我对表中的记录进行审计。有多个列，每个记录状态为 1 个或多个列的更改。
我需要返回一个审核结果，其中返回模式为：column（id、alias 或 name）、以前的值、新的价值等
问题在于，每条新记录可能有多个包含更改数据的列。同时可审计列的数量是 5，因此可以“硬编码”那里的名称和更改验证。

那么是否有可能以缩短的方式编写这样的查询，而不仅仅是使用 UNIONS 并对每一列进行 SELECT 查询并检查更改？

假设有一个带有列的表：

id, datetime value, int value, varchar value.

如果我有 2 条记录的数据更改如下：

id1, value1, value1, value1
id1, value2, value1, value2

那么我期待这样的审计结果：

id1, value1 as oldvalue, value2 as newvalue, column2name as columnname
id1, value1 as oldvalue, value2 as newvalue, column4name as columnname

【问题讨论】：

您不打算同时显示更改时间（或至少显示一些指示更改顺序的顺序值）吗？
另一个问题可能是审计列的类型。如果它们不同，您可能需要将它们全部转换为字符串，以便能够在相同的列中输出它们（即previous value 和new value）。
是的，我必须返回审计发生的日期，是的，我必须将数据转换为 varchar。
MS SQL 2008，但我不会使用 ms sql server 审计。我应该使用现有数据即时执行此操作。
我只询问了版本，因为我想知道您可以使用哪些 Transact-SQL 功能。

标签： sql sql-server sql-server-2008

【解决方案1】：

如果我没有错过任何东西：

WITH ranked AS (
  SELECT
    ChangeDate,
    ColPK,
    Col1,
    Col2,
    Col3,
    Col4,
    Col5,
    OverallRank = ROW_NUMBER() OVER (PARTITION BY ColPK       ORDER BY ChangeDate),
    Col1Rank    = ROW_NUMBER() OVER (PARTITION BY ColPK, Col1 ORDER BY ChangeDate),
    Col2Rank    = ROW_NUMBER() OVER (PARTITION BY ColPK, Col2 ORDER BY ChangeDate),
    Col3Rank    = ROW_NUMBER() OVER (PARTITION BY ColPK, Col3 ORDER BY ChangeDate),
    Col4Rank    = ROW_NUMBER() OVER (PARTITION BY ColPK, Col4 ORDER BY ChangeDate),
    Col5Rank    = ROW_NUMBER() OVER (PARTITION BY ColPK, Col5 ORDER BY ChangeDate)
  FROM AuditTable
)
, ranked2 AS (
  SELECT
    ChangeDate,
    ColPK,
    Col1,
    Col2,
    Col3,
    Col4,
    Col5,
    Col1Group = RANK() OVER (PARTITION BY ColPK, Col1 ORDER BY OverallRank - Col1Rank),
    Col2Group = RANK() OVER (PARTITION BY ColPK, Col2 ORDER BY OverallRank - Col2Rank),
    Col3Group = RANK() OVER (PARTITION BY ColPK, Col3 ORDER BY OverallRank - Col3Rank),
    Col4Group = RANK() OVER (PARTITION BY ColPK, Col4 ORDER BY OverallRank - Col4Rank),
    Col5Group = RANK() OVER (PARTITION BY ColPK, Col5 ORDER BY OverallRank - Col5Rank),
    Col1Rank = ROW_NUMBER() OVER (PARTITION BY ColPK, Col1, OverallRank - Col1Rank ORDER BY ChangeDate),
    Col2Rank = ROW_NUMBER() OVER (PARTITION BY ColPK, Col2, OverallRank - Col2Rank ORDER BY ChangeDate),
    Col3Rank = ROW_NUMBER() OVER (PARTITION BY ColPK, Col3, OverallRank - Col3Rank ORDER BY ChangeDate),
    Col4Rank = ROW_NUMBER() OVER (PARTITION BY ColPK, Col4, OverallRank - Col4Rank ORDER BY ChangeDate),
    Col5Rank = ROW_NUMBER() OVER (PARTITION BY ColPK, Col5, OverallRank - Col5Rank ORDER BY ChangeDate)
  FROM ranked
),
unpivoted AS (
  SELECT
    r.ChangeTime,
    r.ColPK,
    x.ColName,
    ColRank = CASE x.Colname
      WHEN 'Col1' THEN Col1Group
      WHEN 'Col2' THEN Col2Group
      WHEN 'Col3' THEN Col3Group
      WHEN 'Col4' THEN Col4Group
      WHEN 'Col5' THEN Col5Group
    END,
    Value = CASE x.Colname
      WHEN 'Col1' THEN CONVERT(nvarchar(100), r.Col1)
      WHEN 'Col2' THEN CONVERT(nvarchar(100), r.Col2)
      WHEN 'Col3' THEN CONVERT(nvarchar(100), r.Col3)
      WHEN 'Col4' THEN CONVERT(nvarchar(100), r.Col4)
      WHEN 'Col5' THEN CONVERT(nvarchar(100), r.Col5)
    END
  FROM ranked2 r
    INNER JOIN (VALUES ('Col1'), ('Col2'), ('Col3'), ('Col4'), ('Col5')) x (ColName)
      ON x.ColName = 'Col1' AND Col1Rank = 1
      OR x.ColName = 'Col2' AND Col2Rank = 1
      OR x.ColName = 'Col3' AND Col3Rank = 1
      OR x.ColName = 'Col4' AND Col4Rank = 1
      OR x.ColName = 'Col5' AND Col5Rank = 1
)
SELECT
  new.ChangeTime,
  new.ColPK,
  new.ColName,
  old.Value AS OldValue,
  new.Value AS NewValue
FROM unpivoted new
  LEFT JOIN unpivoted old
    ON new.ColPK   = old.ColPK
   AND new.ColName = old.ColName
   AND new.ColRank = old.ColRank + 1

基本上，这个想法是对相同值的连续组进行排名，并选择每个值的第一次出现。这是针对其值正在被审计的每一列完成的，并且这些列在该过程中是非透视的。之后，unpivoted row set 加入到自身，即对于每个 PK 和列名，每行都匹配其前任（基于排名）以获得最终结果集的同一行中的旧值。

【讨论】：

【解决方案2】：

这是一个更简单的查询，它产生相同的期望结果，并且更容易修改以适应不同数量的列或更改列名称，因为唯一的区别是 PK 列 + 每个非单行-CROSS APPLY 中的-PK 列。我必须添加一个ChangeDate 列——没有它，就无法知道插入审计表的行的顺序。

WITH ColValues AS (
   SELECT
      Grp = Row_Number() OVER (
         PARTITION BY H.OrderID, U.ColName ORDER BY H.ChangeDate ASC, X.Which
      ) / 2,
      H.OrderID,
      H.ChangeDate,
      U.*,
      X.Which
   FROM
      dbo.OrderHistory H
      CROSS APPLY (VALUES
         ('DeliveryDate', Convert(varchar(1000), DeliveryDate, 121)),
         ('Quantity', Convert(varchar(1000), Quantity)),
         ('SpecialNotes', Convert(varchar(1000), SpecialNotes))
      ) U (ColName, Value)
      CROSS JOIN (VALUES (1), (2)) X (Which)
)
SELECT
   V.OrderID,
   V.ColName,
   DateChanged = Max(V.ChangeDate),
   OldValue = Max(F.Value),
   NewValue = Max(T.Value)
FROM
   ColValues V
   OUTER APPLY (SELECT V.ColName, V.Value WHERE V.Which = 2) F
   OUTER APPLY (SELECT V.ColName, V.Value WHERE V.Which = 1) T
GROUP BY
   V.OrderID,
   V.ColName,
   V.Grp
HAVING
   Count(*) = 2
   AND EXISTS (
      SELECT Max(F.Value)
      EXCEPT SELECT Max(T.Value)
   )
;

See a live demo of this query at SQL Fiddle.

在 SQL 2012 中，使用 LEAD 或 LAG 分析函数可以更好地解决此问题。我的查询中的CROSS JOIN 和Row_Number 通过复制每一行并将这些复制的行成对分配到它们自己的组中来模拟这一点（其中每个组有两行代表相邻的审计历史记录行）。然后通过策略性地使用聚合，我们可以处理分组对以选择和比较它们的值。

另外，我最初使用UNPIVOT 编写了查询，但可惜它没有保留NULL——在我看来，这是Microsoft 的严重疏忽。如果需要，开发人员可以很容易地添加删除 NULL 的条件，但是当希望保留 NULL 时，根本无法使用 UNPIVOT 的方式。具有讽刺意味的是，使用CROSS APPLY 转换为 UNPIVOT，生成的代码更紧凑，并且缩短了 2 行代码——现在转换和反透视只需一步而不是 2 步。

我的样本数据是：

ChangeDate              OrderID DeliveryDate            Quantity SpecialNotes
----------------------- ------- ----------------------- -------- ----------------------------------------------------
2013-03-01 11:28:00.000 1       2013-04-01 00:00:00.000 25       NULL
2013-03-01 11:56:00.000 1       2013-04-01 00:00:00.000 30       NULL
2013-03-05 10:18:00.000 1       2013-04-02 00:00:00.000 30       Customer called to ask for delivery date adjustment.
2013-03-01 11:37:00.000 2       2013-03-05 00:00:00.000 17       NULL

结果行集：

OrderID ColName      DateChanged             OldValue                NewValue
------- ------------ ----------------------- ----------------------- ---------------------------------------------------
1       DeliveryDate 2013-03-05 10:18:00.000 2013-04-01 00:00:00.000 2013-04-02 00:00:00.000
1       Quantity     2013-03-01 11:56:00.000 25                      30
1       SpecialNotes 2013-03-05 10:18:00.000 NULL                    Customer called to ask for delivery date adjustment.

注意：由于我的查询只有一个排名函数并且没有JOINs，因此即使在非常大的表中也能表现得非常好——也许比使用JOIN 的解决方案要好几个数量级。没有支撑指数。审计表最好在PK, ChangeDate 上有一个聚集索引。

【讨论】：