数据库内核月报－ 2016/11 - MSSQL · 最佳实战 · 巧用COLUMNS_UPDATED获取数据变更 - 《数据库内核月报》

业务场景
- 全量更新
- 全量+增量更新
MSSQL获取数据变更
总结

业务场景

在平时与数据库打交道的过程中，我们经常会有这样的疑惑：如何快速的获取数据变更记录呢？举个例子，搜索引擎要为外部客人提供快速准确的商品信息搜索功能，那么当有新的商品数据变更后，搜索引擎如何快速的发现这些新的变更数据呢？我们常见的两种做法：

全量更新

这种方法最为简单直接，反正不管三七二十一，搜索引擎每次全量拉取商品信息表所有数据，然后创建搜索索引，提供给外部客人查询。这种方法实现起来的确最为简单，当然同时也具有非常明显的缺点：

浪费资源： 假如商品的变更频率为20%，那么剩下的80%商品实际上是不需要更新的。换句话说全量更新会浪费掉80%的系统资源(IO/CPU/Memory)来做无用功。
耗时严重： 由于获取的是表的全量数据，所以全量更新大大增加了数据获取阶段和搜索索引生成阶段锁的概率，加之浪费资源做无用功，最终导致时间消耗大大拉长。
数据更新时效性差： 由于耗时严重，所以导致数据更新不及时，时效性差，随着商品量的不断扩大，这种时效性会越来越差，最终导致客户抱怨。

全量+增量更新

针对全量更新的种种“罪行”，我们可以有针对性的采用全量+增量更新的方式来有效解决。这种方法的思路是，我们可以周期性的做全量更新，比如每天或者每周，然后在两个全量更新周期之间，我们采用增量更新的方式来覆盖新的数据变更，比如每小时或者每分钟。增量更新问题的关键在于如何获取数据变更记录，让我们来看看关系型数据库MSSQL Server是如何提供解决方法的。

MSSQL获取数据变更

MSSQL Server提供了一个函数，名为COLUMNS_UPDATED可以解决这个问题。先让我们来看看微软官方的解释：返回 varbinary 位模式，它指示表或视图中插入或更新了哪些列。官方文档的解释非常的抽象，如果想要使用这个函数来获取数据变更记录，我们需要踩过很多坑，突破很多点，这也是这篇文章的价值。

COLUMNS_UPDATED

首先，我们来看看这个函数表达的含义。假如某张表有8个字段，那么COLUMNS_UPDATED使用一个byte，八个bit来表示哪些列发生了数据变更，表示方法如下：

Col_id	8	7	6	5	4	3	2	1
Bit	7	6	5	4	3	2	1	0
Value	128	64	32	16	8	4	2	1

Col_id：表字段顺序ID

Bit：bit位顺序，从0开始

Value：2的bit次方

当某些列被更新后，COLUMNS_UPDATED函数会返回varbinary位模式（varbinary位模式是什么？可以理解为所有列Value的SUM值的二进制格式）。比如：当第二列和第四列被更新，那么COLUMNS_UPDATED的varbinary位模式是2 + 8 = 10。来看一个具体的例子。


1. use tempdb
2. GO
3. IF EXISTS(SELECT TOP 1 1
4. FROM sys.tables
5. WHERE name = 'employeeData')
6. DROP TABLE employeeData;
7. GO

9. CREATE TABLE dbo.employeeData (
10. col1 int identity(1,1) not null,
11. col2 int NOT NULL,
12. col3 int NOT NULL,
13. col4 int NOT NULL,
14. col5 int NOT NULL,
15. col6 int NOT NULL,
16. col7 int NOT NULL constraint uni unique,
17. col8 int NOT NULL,
18. );
19. GO

21. CREATE TRIGGER dbo.Trg_UID_employeeData
22. ON dbo.employeeData
23. AFTER UPDATE,INSERT,DELETE
24. AS
25. BEGIN
26. declare
27. @table_id int = 0
28. ;
29. select top 1
30. @table_id = parent_id
31. from sys.triggers with(nolock)
32. where object_id = @@procid;

34. select updated_columns =
35. stuff
36. (replace(
37. replace(
38. (
39. select column_name = quotename(name)
40. from sys.columns with(Nolock)
41. where object_id = @table_id
42. and CONVERT(VARBINARY,COLUMNS_UPDATED()) & POWER(2, column_id - 1)  = POWER(2, column_id - 1)
43. order by column_id asc
44. for xml path('')
45. )
46. ,'<column_name>',',')
47. ,'</column_name>','')
48. ,1,1,'')
49. ,columns_updated_value = cast(COLUMNS_UPDATED() as int)
50. END
51. GO

53. --test DML actions
54. --INSERT
55. INSERT INTO dbo.employeeData
56. VALUES ( 2, 3, 4, 5, 6, 7, 8);
57. GO

59. --UPDATE
60. UPDATE A
61. SET col2 = col2 + 10
62. ,col4 = col4 + 11
63. FROM dbo.employeeData AS A

65. --DELETE
66. delete from dbo.employeeData

结果如下：

踩过的坑1: INT数据类型溢出

注意上面的代码POWER(2, column_id - 1)，返回的应该是一个INT数据类型的值。在MSSQL SQL Server中INT类型使用4个字节来存储，也就是32bit，换句话说，当表的字段列个数达到32时，这个POWER操作会导致INT数据类型溢出而报告异常。当我们将上面的表字段加到32个后，INSERT和UPDATE操作会导致TRIGGER报告如下错误：


1. Msg 232, Level 16, State 3, Procedure Trg_UID_employeeData, Line 15
2. Arithmetic overflow error for type int, value = 2147483648.000000.
3. The statement has been terminated.
4. Msg 232, Level 16, State 3, Procedure Trg_UID_employeeData, Line 15
5. Arithmetic overflow error for type int, value = 2147483648.000000.
6. The statement has been terminated.

踩过的坑2：BIGINT数据类型溢出

关于这个问题，在没有完美的解决方法之前，很长一段时间，我们强制将POWER转化为BIGINT数据类型来暂时突破32个字段数量限制。但是，这个坑原理和上面一样，仅仅是将字段数量从32个扩大到64个。方法如下：


1. ...
2. and CONVERT(VARBINARY,COLUMNS_UPDATED()) & POWER(cast(2 as bigint), column_id - 1)  = POWER(cast(2 as bigint), column_id - 1)
3. ...

如何完美的解决上面两个坑，我们先暂时留个悬念。

庖丁解牛

让我们回到最原始的需求，对于DML操作，不外乎三种，即INSERT，UPDATE和DELETE。我们的Trigger必须具备识别这三种操作类型的能力。

INSERT：Trigger需要具备识别表数据行唯一标识(RID)的能力（通常是主键），然后通过RID反过来查询正式表即可。

UPDATE：Trigger需要具备识别哪些字段被更新的能力，然后通过RID获取这些被更新的字段的值。

DELETE：Trigger获取到数据行唯一标识即可，通过RID删除对应的行。

综合了所有这些分析以后，我们可以使用如下的TRIGGER来捕获数据变更。


1. use tempdb
2. GO
3. --create table to save changed data.
4. if object_id('dbo.triggeredDataLog', 'U') is not null
5. drop table dbo.triggeredDataLog
6. GO
7. create table dbo.triggeredDataLog(
8. rowid bigint identity(1,1) not null primary key,
9. database_name sysname not null,
10. schame_name sysname not null,
11. table_object_name sysname not null,
12. operation char(1) not null,
13. RID nvarchar(1000) not null,
14. updated_columns nvarchar(max) null,
15. indate datetime not null default (getdate()),
16. intime timestamp not null,
17. )

19. IF EXISTS(SELECT TABLE_NAME FROM INFORMATION_SCHEMA.TABLES
20. WHERE TABLE_NAME = 'employeeData')
21. DROP TABLE employeeData;
22. GO

24. --create table for testing.
25. CREATE TABLE dbo.employeeData (
26. id int identity(1,1) not null,
27. c1 int null,
28. c2 int null,
29. c3 int null,
30. c4 int null,
31. c5 int null,
32. c6 int null,
33. c7 int null,
34. c8 int null,
35. c9 int null,
36. c10 int null,
37. c11 int null,
38. c12 int null,
39. c13 int null,
40. c14 int null,
41. c15 int null,
42. c16 int null,
43. c17 int null,
44. c18 int null,
45. c19 int null,
46. c20 int null,
47. c21 int null,
48. c22 int null,
49. c23 int null,
50. c24 int null,
51. c25 int null,
52. c26 int null,
53. c27 int null,
54. c28 int null,
55. c29 int null,
56. c30 int null,
57. c31 int null,
58. c32 int null,
59. c33 int null,
60. c34 int null,
61. c35 int null,
62. c36 int null,
63. c37 int null,
64. c38 int null,
65. c39 int null,
66. c40 int null,
67. c41 int null,
68. c42 int null,
69. c43 int null,
70. c44 int null,
71. c45 int null,
72. c46 int null,
73. c47 int null,
74. c48 int null,
75. c49 int null,
76. c50 int null,
77. c51 int null,
78. c52 int null,
79. c53 int null,
80. c54 int null,
81. c55 int null,
82. c56 int null,
83. c57 int null,
84. c58 int null,
85. c59 int null,
86. c60 int null,
87. c61 int null,
88. c62 int null,
89. c63 int null,
90. c64 int null,
91. c65 int null,
92. c66 int null,
93. c67 int null,
94. c68 int null,
95. c69 int null
96. );
97. GO

101. CREATE TRIGGER dbo.Trg_UID_employeeData
102. ON dbo.employeeData
103. AFTER UPDATE,INSERT,DELETE
104. AS
105. BEGIN
106. SET NOCOUNT ON;
107. --=======================================
108. -- get DML Action (INSERT,UPDATE,DELETE)
109. DECLARE
110. @OperationType CHAR(1)
111. ,@table_id int = 0
112. ;
113. select top 1
114. @OperationType = 'D'
115. ,@table_id = parent_id
116. from sys.triggers with(nolock)
117. where object_id = @@procid
118. ;

120. --get operation type:
121. --record in inserted & deleted, that means UPDATE DML operation
122. --record in inserted but not in deleted, that means INSERT DML operation
123. --by default, we set operation type as DELETE DML operation
124. IF EXISTS (SELECT TOP 1 1 FROM inserted)
125. BEGIN
126. IF EXISTS (SELECT TOP 1 1 FROM deleted)
127. BEGIN
128. SET @OperationType = 'U'; --UPDATE
129. END
130. ELSE
131. SET @OperationType = 'I'; --INSERT
132. END

134. -- end of getting DML Action
135. --=======================================
136. -- we need to konw PK column(s) or identity column or unqiue column
137. -- table exists PK
138. declare
139. @tb_unique_cols table(
140. column_name sysname not null
141. ,data_type sysname not null)

143. IF EXISTS( --primary key
144. select * from sys.indexes WITH(NOLOCK)
145. where object_id = @table_id
146. and is_primary_key = 1
147. )
148. BEGIN
149. INSERT INTO @tb_unique_cols
150. SELECT
151. column_name = col.name
152. ,data_type = ty.name
153. FROM    sys.indexes AS i with(NOLOCK)
154. INNER JOIN sys.index_columns AS ic  with(NOLOCK)
155. ON  i.OBJECT_ID = ic.OBJECT_ID
156. AND i.index_id = ic.index_id
157. INNER JOIN sys.columns AS col with(NOLOCK)
158. ON i.object_id = col.object_id
159. INNER JOIN sys.types as ty with(NOLOCK)
160. ON col.user_type_id = ty.user_type_id
161. WHERE   i.is_primary_key = 1
162. and i.object_id = @table_id
163. and ic.column_id = col.column_id
164. END
165. ELSE IF EXISTS(    --table doesn't have primary key but table exists identity
166. select * from sys.columns
167. where object_id = @table_id
168. and is_identity = 1
169. )
170. BEGIN
171. INSERT INTO @tb_unique_cols
172. select column_name = col.name,data_type = ty.name
173. from sys.columns as col with(NOLOCK)
174. INNER JOIN sys.types as ty with(NOLOCK)
175. ON col.user_type_id = ty.user_type_id
176. where col.object_id = @table_id
177. and col.is_identity = 1
178. END
179. ELSE IF EXISTS(    --table doesn't have primary key/indentity but table has unique index or constraint
180. select * from sys.indexes with(NOLOCK)
181. where object_id = @table_id
182. and is_unique = 1
183. )
184. BEGIN
185. INSERT INTO @tb_unique_cols
186. SELECT  TOP 1 column_name = col.name
187. ,data_type = ty.name
188. FROM    sys.indexes AS i  with(NOLOCK)
189. INNER JOIN sys.index_columns AS ic  with(NOLOCK)
190. ON  i.OBJECT_ID = ic.OBJECT_ID
191. AND i.index_id = ic.index_id
192. INNER JOIN sys.columns AS col with(NOLOCK)
193. ON i.object_id = col.object_id
194. and col.column_id = ic.column_id
195. INNER JOIN sys.types as ty with(NOLOCK)
196. ON col.user_type_id = ty.user_type_id
197. WHERE   i.is_unique = 1
198. and i.object_id = @table_id
199. END

201. --=======================================
202. --get PK set: [pk1] = 1 and [pk2] = 'ABSDEF' and [pk3] = 'Jul 29 2016  5:04PM'
203. declare
204. @unique_cols_list nvarchar(max)
205. ,@sql nvarchar(max)
206. ,@RID nvarchar(max)
207. ,@database_name sysname
208. ,@schema_name sysname
209. ,@table_object_name sysname
210. ;
211. select @unique_cols_list = '''' +
212. stuff
213. (replace(
214. replace(
215. (
216. select column_name = N'+ ' + quotename( N' and ' + quotename(column_name)+ N' = ', '''') + N'+ ' +
217. case
218. when data_type in ('char','nchar','varchar','nvarchar','date','datetime','datetime2','smalldatetime') then N'quotename('
219. else  ''
220. end  +'cast('+quotename(column_name) +' as varchar)' +
221. case
222. when data_type in ('char','nchar','varchar','nvarchar','date','datetime','datetime2','smalldatetime') then N','''''''')'
223. else ''
224. end

226. from @tb_unique_cols
227. for xml path('')
228. )
229. ,'<column_name>','')
230. ,'</column_name>','')
231. ,1,8,'')
232. ,@database_name = db_name()
233. ,@schema_name = schema_name(schema_id)
234. ,@table_object_name = object_name(object_id)
235. from sys.tables
236. where object_id = @table_id
237. --end get PK set

239. --end of table PK/identity/unique generation
240. --=======================================
241. -- recording the DML into log

243. IF @OperationType = 'I'    --INSERT
244. BEGIN
245. IF EXISTS(select TOP 1 1 from inserted)
246. BEGIN
247. select * into #inserted from inserted
248. set
249. @sql = N'SELECT @RID = '+ @unique_cols_list + N' FROM #inserted'
250. ;

252. exec sys.sp_executesql @sql
253. ,N'@RID nvarchar(max) output'
254. ,@RID = @RID output
255. ;

257. --select @sql,@RID
258. INSERT INTO dbo.triggeredDataLog(database_name,schame_name,table_object_name,operation,RID)
259. select @database_name,@schema_name,@table_object_name,@OperationType, @RID
260. END
261. END
262. ELSE IF @OperationType = 'U' --UPDATE
263. BEGIN
264. --we need to konw PK column(s) & updated columns
265. IF EXISTS(select TOP 1 1 from deleted)
266. BEGIN
267. /*start
268. get updated columns
269. */
270. DECLARE
271. @Columns_Updated NVARCHAR(max)
272. ,@maxByteCU INT
273. ,@curByteCU INT
274. ,@cByte INT
275. ,@curBit INT
276. ,@maxBit INT
277. ;

279. SELECT
280. @maxByteCU = DATALENGTH(COLUMNS_UPDATED())
281. ,@Columns_Updated = N''
282. ,@curByteCU = 1

284. WHILE @curByteCU <= @maxByteCU
285. BEGIN
286. SELECT @cByte = SUBSTRING(COLUMNS_UPDATED(), @curByteCU, 1)
287. ,@curBit = 1
288. ,@maxBit = 8
289. ;

291. WHILE @curBit <= @maxBit
292. BEGIN
293. IF CONVERT(BIT, @cByte & POWER(2,@curBit - 1)) <> 0
294. --SET @Columns_Updated = @Columns_Updated + '[' + CONVERT(VARCHAR, 8 * (@curByteCU - 1) + @curBit) + ']'
295. select @Columns_Updated = @Columns_Updated + QUOTENAME(name) + ','
296. from sys.columns with(Nolock)
297. where object_id = @table_id
298. and column_id = 8 * (@curByteCU - 1) + @curBit

300. SET @curBit = @curBit + 1
301. END
302. SET @curByteCU = @curByteCU + 1
303. END
304. /*end
305. get updated columns
306. */

308. select * into #deleted from deleted

310. set
311. @sql = N'SELECT @RID = '+ @unique_cols_list + N' FROM #deleted'
312. ;

314. exec sys.sp_executesql @sql
315. ,N'@RID nvarchar(max) output'
316. ,@RID = @RID output
317. ;
318. INSERT INTO dbo.triggeredDataLog(database_name,schame_name,table_object_name,operation,RID,updated_columns)
319. select @database_name,@schema_name,@table_object_name,@OperationType, @RID,left(@Columns_Updated,len(@Columns_Updated) - 1)

321. END
322. END
323. ELSE    --DELETE
324. BEGIN
325. --we need to konw PK column(s)
326. IF EXISTS(select TOP 1 1 from deleted)
327. BEGIN
328. select * into #deleted1 from deleted

330. set
331. @sql = N'SELECT @RID = '+ @unique_cols_list + N' FROM #deleted1'
332. ;

334. exec sys.sp_executesql @sql
335. ,N'@RID nvarchar(max) output'
336. ,@RID = @RID output
337. ;
338. INSERT INTO dbo.triggeredDataLog(database_name,schame_name,table_object_name,operation,RID)
339. select @database_name,@schema_name,@table_object_name,@OperationType, @RID
340. END
341. END
342. END
343. GO

346. --=======================================
347. -- table just has identity column
348. -- Testing INSERT
349. INSERT INTO dbo.employeeData(c1,c2,c3,c4)
350. VALUES(1,2,3,4)

352. --Testing UPDATE
353. UPDATE TOP(1) A
354. SET c64 = 64
355. ,c65 = 65
356. FROM dbo.employeeData AS A

358. --Testing DELETE
359. DELETE TOP (1) A
360. FROM dbo.employeeData AS A

362. --=======================================
363. -- table has unique constraint

365. ALTER TABLE dbo.employeeData
366. DROP COLUMN ID;

368. ALTER TABLE dbo.employeeData ADD
369. c70 int NOT NULL constraint uni_c70 unique
370. GO

372. -- Testing INSERT
373. INSERT INTO dbo.employeeData(c1,c2,c3,c4,c70)
374. VALUES(1,2,3,4,70)

376. --Testing UPDATE
377. UPDATE TOP(1) A
378. SET c64 = 64
379. ,c65 = 65
380. FROM dbo.employeeData AS A

382. --Testing DELETE
383. DELETE TOP (1) A
384. FROM dbo.employeeData AS A
385. --=======================================
386. -- table has primary key
387. ALTER TABLE dbo.employeeData ADD
388. pk1 int NOT NULL,
389. pk2 varchar(100) not null,
390. pk3 datetime not null default(getdate());

392. ALTER TABLE dbo.employeeData ADD
393. CONSTRAINT pk primary key(pk1,pk2,pk3)
394. GO

396. -- Testing INSERT
397. INSERT INTO dbo.employeeData(pk1,pk2,pk3,c70)
398. VALUES(1,2,GETDATE(),70)

400. --Testing UPDATE
401. UPDATE TOP(1) A
402. SET c64 = 64
403. ,c65 = 65
404. FROM dbo.employeeData AS A

406. --Testing DELETE
407. DELETE TOP (1) A
408. FROM dbo.employeeData AS A
409. GO

411. select * from dbo.triggeredDataLog with(NOLOCK) order by intime asc

结果分析

最后一条查询语句结果如下截图：

Rowid 1-3:表无主键，但存在IDENTITY属性列的情况，RID为IDENTITY属性列的值，我们抓取到的RID和Updated_columns

Rowid 4-6:表无主键，但存在UNIQUE约束的情况，RID为UNIQUE列的值，取到的RID和Updated_columns

Rowid 7-9:表有主键，这里是更加复杂的联合主键，RID为联合主键的值，取到的RID和Updated_columns

在本例的表字段个数超过了64个，达到73个，我们是采用循环获取的方式来踩过坑1和2，具体代码268行到307行。

总结

到目前为止，我们的搜索引擎只需要从dbo.triggeredDataLog表中获取数据变更RID和相应发生了变化的字段Updated_columns，而不需要从正式表中整个拉取全量数据，节约了数据库系统开销，增加了搜索索引创建的时效性，提高了客户体验。

注意：

这里需要特别提醒，正式表dbo.employeeData上千万不要使用TRUNCATE TABLE的操作，因为TRUNCATE动作无法激活触发器。


1. --forbidden action
2. TRUNCATE TABLE dbo.employeeData;

原文：http://mysql.taobao.org/monthly/2016/11/08/