用 hstore 稽核異動 • PostgreSQL 的藝術

hstore 擴充套件實作了一種資料型別，能在單一 PostgreSQL 值中儲存一組鍵值對（key/value pairs），適合「屬性很多但很少被查看的資料列」或半結構化資料。鍵與值都只是文字字串。

可以說 hstore 是 PostgreSQL JSON 支援的前身，兩者涵蓋部分相同的使用情境。主要差異在於：

hstore 只支援一種資料型別——text。
hstore 複合值是扁平的字典，不支援巢狀。

即便如此，hstore 在某些情境依然非常好用。本章示範如何用 hstore 以通用方式稽核資料異動。

hstore 入門#

第一步當然是啟用擴充套件：

create extension hstore;

之後就能建立 hstore 值，並用箭頭運算子 -> 取出指定鍵的值——傳入單一鍵取得純量值，傳入 array['a', 'c'] 則一次取出多個鍵的值：

select kv,
       kv->'a' as "kv -> a",
       kv-> array['a', 'c'] as "kv -> [a, c]"
  from (
         values ('a=>1,a=>2'::hstore),
                ('a=>5,c=>10')
       )
       as t(kv);

          kv         │ kv -> a │ kv -> [a, c]
═════════════════════╪═════════╪══════════════
 "a"=>"1"            │ 1       │ {1,NULL}
 "a"=>"5", "c"=>"10" │ 5       │ {5,10}
(2 rows)

從結果可見，hstore 的鍵與值全都是文字值。

比較兩個 hstore#

hstore 實作了 - 運算子，文件說明是「從左運算元刪除相符的鍵值對」：

select     'f1 => a, f2 => x'::hstore
         - 'f1 => b, f2 => x'::hstore
         as diff;

   diff
═══════════
 "f1"=>"a"
(1 row)

這個差集運算子正是稽核觸發器（trigger）要用的核心工具——它產生的格式非常適合用來理解「到底改了什麼」。

用觸發器稽核異動#

先做一些準備：

追蹤對象是前一章處理過的 MoMA 收藏資料，被稽核的表為 moma.artist。
異動記錄在 moma.audit 表中，其定義相當通用。
在 moma.artist 上安裝 trigger，捕捉所有異動，把更新前後的資料列版本寫進 moma.audit。
資料列以 hstore 格式記錄，這種表示法非常有彈性：可以同時追蹤多張表，甚至同一張表經 ALTER TABLE 改過 schema 之後也照樣適用。

想法是：每次 moma.artist 被更新，就在稽核表新增一列，存放異動前後的 hstore 表示：

begin;

create table moma.audit
  (
    change_date timestamptz default now(),
    before      hstore,
    after       hstore
  );

commit;

接著是 hstore 稽核觸發器：

begin;

create function moma.audit()
   returns trigger
  language plpgsql
as $$
begin
  INSERT INTO audit(before, after)
       SELECT hstore(old), hstore(new);
  return new;
end;
$$;

create trigger audit
       after update on moma.artist
           for each row
  execute procedure audit();

commit;

稽核表的定義完全不含 moma.artist 專屬的細節，所以同一個 trigger 可以掛到任何其他資料表上。這麼做時，就需要再加上 schema_name 與 table_name 兩個欄位記錄異動來源；在觸發器程序內，這些資訊可從 TG_TABLE_SCHEMA 與 TG_TABLE_NAME 變數取得。想強化本例的觸發器程式碼，可閱讀官方文件〈PL/pgSQL Trigger Procedures〉一章。

延伸範例：含來源欄位的通用稽核表定義

begin;

create table moma.audit
  (
    change_date timestamptz default now(),
    schema_name name,
    table_name  name,
    before      hstore,
    after       hstore
  );

commit;

測試稽核觸發器#

準備就緒後實際測試：把新一批 CSV 匯入暫存表，更新有變動的資料列、插入新資料列（與前一章相同的批次更新手法）：

begin;

create temp table batch
  (
        like moma.artist
    including all
  )
  on commit drop;

\copy batch from 'artists/artists.2017-07-01.csv' with csv header delimiter ','

with upd as
(
        update moma.artist
           set (name, bio, nationality, gender, begin, "end", wiki_qid, ulan)
             = (batch.name, batch.bio, batch.nationality,
                batch.gender, batch.begin, batch."end",
                batch.wiki_qid, batch.ulan)
          from batch
         where batch.constituentid = artist.constituentid
           and (artist.name, artist.bio, artist.nationality,
                artist.gender, artist.begin, artist."end",
                artist.wiki_qid, artist.ulan)
            <> (batch.name, batch.bio, batch.nationality,
                batch.gender, batch.begin, batch."end",
                batch.wiki_qid, batch.ulan)
     returning artist.constituentid
),
       ins as
(
       insert into moma.artist
            select constituentid, name, bio, nationality,
                   gender, begin, "end", wiki_qid, ulan
              from batch
             where not exists
                   (
                         select 1
                           from moma.artist
                          where artist.constituentid = batch.constituentid
                   )
       on conflict (constituentid) do nothing
         returning artist.constituentid
)
select (select count(*) from upd) as updates,
       (select count(*) from ins) as inserts;

commit;

執行結果：匯入 15,226 列，其中 52 筆更新、61 筆插入。

多虧稽核觸發器，現在可以直接看「改了什麼」：

select (before -> 'constituentid')::integer as id,
       after - before as diff
  from moma.audit
  limit 15;

   id  │                               diff
═══════╪═════════════════════════════════════════════════════════════════
   546 │ "bio"=>"American, born England. 1906–1994"
   570 │ "bio"=>"American, 1946–2016"
   920 │ "bio"=>"American, born Switzerland. 1907–1988", "end"=>"1988"
  1372 │ "bio"=>"Belgian, 1901–1986", "end"=>"1986", "name"=>"Suzanne va…
       │…n Damme", "begin"=>"1901", "nationality"=>"Belgian"
  1669 │ "name"=>"Dušan Džamonja"
  1754 │ "name"=>"Erró (Gudmundur Gudmundsson)"
  ...
(15 rows)

從 hstore 還原成一般資料列#

hstore 能用 hstore() 函式把 record 轉成 hstore，也能用 populate_record() 轉回來。以下範例利用這個強大的函式，找出所有藝術家名字被改過的紀錄，並顯示異動時間、舊名與新名：

select audit.change_date::date,
       artist.name as "current name",
       before.name as "previous name"

  from      moma.artist
       join moma.audit
         on (audit.before->'constituentid')::integer
          = artist.constituentid,
       populate_record(NULL::moma.artist, before) as before

where artist.name <> before.name;

查詢先從稽核表取出 constituentid 與 artist 表 join，得到：

 change_date │         current name         │        previous name
═════════════╪══════════════════════════════╪══════════════════════════════
 2018-08-25  │ Suzanne van Damme            │ Elisabeth van Damme
 2018-08-25  │ Dušan Džamonja               │ Dusan Dzamonja
 2018-08-25  │ Erró (Gudmundur Gudmundsson) │ Erro (Gudmundur Gudmundsson)
 2018-08-25  │ Nikos Hadjikyriakos-Ghika    │ Nikos HadjiKyriakos-Ghika
 2018-08-25  │ Sam Mendes                   │ Same Mendes
 2018-08-25  │ Tim Berresheim               │ Tim Berrescheim
 2018-08-25  │ Kestutis Nakas               │ Kęstutis Nakas
 2018-08-25  │ Jennifer T. Ley              │ Jennifer Ley
(8 rows)

即使現今的 PostgreSQL 已有 JSON 支援，hstore 仍非常有用：record 與 hstore 互轉的能力是這個擴充獨有的，而它的差集運算子在 JSON 功能集中沒有對應物。