Logical Decoding是9.4里面的一個主要功能，是向最終實現邏輯復制邁出的一大步。簡言之，它的功能是從PG的WAL日志中，讀取數據庫更新信息，然后“翻譯”（Decode）成邏輯的形式，可發送到遠程從庫做數據同步。這個功能還可以用于，DBA在數據庫宕機，并發生主從切換后，檢查原主庫有哪些更新宕機前未同步到從庫，并手動同步來彌補丟失的（已提交）的更新。這里我們探索一下它的使用和實現原理。 **使用** 1）首先需要將 wal_level這個配置參數設置為logical，并保證max_replication_slots至少為1。 2）創建Logical Replication Slot。Logical Decoding利用了Logical Replication Slot來獲取和Decode日志。關于Physical Replication Slot我們在上期中有詳細介紹，而Logical Replication Slot與Physical Replication Slot的數據結構類似。創建一個Logical Replication Slot的命令如下： ~~~ SELECT * FROM pg_create_logical_replication_slot('my_rep_slot', 'test_decoding'); slot_name | xlog_position -------------+--------------- my_rep_slot | 0/7FE68E8 ~~~ ~~~ select * from pg_replication_slots; slot_name | plugin | slot_type | datoid | database | active | xmin | catalog_xmin | restart_lsn ---------------+---------------+-----------+--------+----------+--------+------+--------------+------------- my_rep_slot | test_decoding | logical | 13003 | postgres | f | | 9735 | 0/7FE6828 ~~~ 注意，創建Logical Replication Slot，需要指定一個輸出插件（Output Plugin）。這個插件要提供一些回調函數，用于格式化輸出日志。就是說，內核中的Logical Decoding先讀取WAL日志，將其Decode成一種半成品式的格式（已包含所有有效信息，比如被更新的表名，更新類型，更新前后的數據記錄即Tuple），然后交由輸出插件最終呈現給用戶。這里我們使用了系統自帶的一個插件，即test_decoding。上面的輸出中，創建命令返回的xlog_position的值是當前系統中，最后被寫入磁盤的日志記錄的LSN。此LSN之后的日志，都可以通過Logical Decoding進行解析了。 3）解析日志。 ~~~ --先做一個執行插入操作的事務： begin; insert into test values(2); commit; --使用pg_logical_slot_peek_changes，Decode對應的WAL日志。 SELECT * FROM pg_logical_slot_peek_changes('my_rep_slot', NULL, NULL, 'include-timestamp', 'on', 'include-xids', 'on'); location | xid | data -----------+------+------------------------------------------------ 0/7FE6C80 | 9736 | BEGIN 9736 0/7FE6C80 | 9736 | table public.test: INSERT: col[integer]:2 0/7FE6DE0 | 9736 | COMMIT 9736 (at 2015-03-13 15:03:49.803582+08) ~~~ pg_logical_slot_peek_changes返回數據中的第二行記錄了我們所做的INSERT操作（只有在事務提交后，才能看到這些修改）。而我們通過pg_xlogdump可以看到原來的WAL日志記錄為： ~~~ <PG installdir>/bin/pg_xlogdump -s 0/7FE6C80 -n 1 rmgr: Heap len (rec/tot): 31/ 219, tx: 9736, lsn: 0/07FE6C80, prev 0/07FE6C48, bkp: 1000, desc: insert: rel 1663/13003/16507; tid 0/3 ~~~ 也就是說，Logical Decoding把這條日志，反解析成一個“table public.test: INSERT: col[integer]:2”字符串。其實如果對輸出插件稍作修改，可以直接解析成可執行的SQL語句：“INSERT INTO public.test (col) VALUES(2)“ 那么這是如何做到的呢？下面我們看看其中原理。 **原理** 追蹤一下pg_logical_slot_peek_changes的調用鏈，不難看到Decoding的整個過程。在pg_logical_slot_get_changes_guts中，從restart_lsn（即上次的最后讀取后，剩下的事務中最先開始的事務對應的LSN）開始，先用XLogReadRecord函數（注意，會先從cache里面讀取日志，如果cache里面沒有，則會到磁盤中的日志段里面讀取）獲取一個日志記錄，存入結構體XLogRecord，緊接著用LogicalDecodingProcessRecord做Decode。如此循環，直到讀完日志或到達指定點。 LogicalDecodingProcessRecord是解析日志的關鍵。它在內存中維護一個哈希表（LogicalDecodingContext->reorder->by_txn），存放正在處理的事務信息。在處理每個日志記錄時，如果遇到一個BEGIN操作，就在哈希表中插入相應事務。而只有在遇到COMMIT操作的時候，才會把整個事務的所有語句解析出來（調用ReorderBufferCommit）。這個過程中，它要為每個事務維護一個快照（Snapshot）。每次有事務做COMMIT都要更新一下這個快照。這樣，等到事務COMMIT時，它的快照是最新的，可以用來訪問系統表，得到如relation node id與relation名字之間的對應關系等信息，從而完成Decode。需要說明的是，LogicalDecodingProcessRecord在維護快照時做了優化：因為Decode過程只需要訪問系統表，所以快照中只保留了那些更新了系統表的事務。 另外，Replication Slot的xmin信息會影響系統的Vacuum，使其保留仍然需要的數據版本。而SnapBuildProcessRunningXacts會不斷更新Replication Slot中的xmin信息，避免使Vacuum停滯。 輸出Decode后日志的過程，都在DecodeCommit調用的ReorderBufferCommit函數中。在ReorderBufferCommit中，調用了輸出插件的apply_change等回調函數，會將日志信息打印成我們最終看到的字符串，這樣就完成了Decode。