PostgreSQL源码分析——pg

PostgreSQL源码分析——pg_waldump

PG中有个可以打印WAL日志信息的工具，pg_waldump，对于开发人员来说，掌握该工具是十分有必要的。

用法

具体用法这里不去详解，可见官网pg_waldump文档。

postgres@slpc:~$ pg_waldump --help
pg_waldump decodes and displays PostgreSQL write-ahead logs for debugging.Usage:pg_waldump [OPTION]... [STARTSEG [ENDSEG]]Options:-b, --bkp-details      output detailed information about backup blocks-e, --end=RECPTR       stop reading at WAL location RECPTR-f, --follow           keep retrying after reaching end of WAL-n, --limit=N          number of records to display-p, --path=PATH        directory in which to find log segment files or adirectory with a ./pg_wal that contains such files(default: current directory, ./pg_wal, $PGDATA/pg_wal)-q, --quiet            do not print any output, except for errors-r, --rmgr=RMGR        only show records generated by resource manager RMGR;use --rmgr=list to list valid resource manager names-s, --start=RECPTR     start reading at WAL location RECPTR-t, --timeline=TLI     timeline from which to read log records(default: 1 or the value used in STARTSEG)-V, --version          output version information, then exit-x, --xid=XID          only show records with transaction ID XID-z, --stats[=record]   show statistics instead of records(optionally, show per-record statistics)-?, --help             show this help, then exit

具体的，我们举一个例子：

postgres=# create table t1(a int);
CREATE TABLEpostgres=# select pg_current_wal_lsn();pg_current_wal_lsn 
--------------------0/1696C78
(1 row)
-- 插入一条数据
postgres=# insert into t1 values(1);
INSERT 0 1
postgres=# select pg_current_wal_insert_lsn();pg_current_wal_insert_lsn 
---------------------------0/1696D18
(1 row)
postgres=# select txid_current();txid_current 
--------------735
(1 row)-- 查看WAL日志
postgres@slpc:~/pgsql$ pg_waldump -p pgdata/pg_wal/ -s 0/1696C78 -e 0/1696D18
-- 插入一条数据，堆表， 事务ID 735   
rmgr: Heap        len (rec/tot):     59/    59, tx:        734, lsn: 0/01696C78, prev 0/01696C40, desc: INSERT+INIT off 1 flags 0x00, blkref #0: rel 1663/13010/16384 blk 0 
rmgr: Transaction len (rec/tot):     34/    34, tx:        734, lsn: 0/01696CB8, prev 0/01696C78, desc: COMMIT 2023-09-23 10:58:39.915785 CST
rmgr: Standby     len (rec/tot):     54/    54, tx:          0, lsn: 0/01696CE0, prev 0/01696CB8, desc: RUNNING_XACTS nextXid 735 latestCompletedXid 733 oldestRunningXid 734; 1 xacts: 734-- 设置wal_debug = on
postgres=# set wal_debug = on;
SET
postgres=# select pg_current_wal_lsn();pg_current_wal_lsn 
--------------------0/1696E28
(1 row)postgres=# insert into t1 values(2);
LOG:  INSERT @ 0/1696EC0:  - Heap/INSERT: off 2 flags 0x00
LOG:  INSERT @ 0/1696EE8:  - Transaction/COMMIT: 2023-09-23 11:27:25.399199+08
LOG:  xlog flush request 0/1696EE8; write 0/1696E28; flush 0/1696E28
INSERT 0 1
postgres=# select pg_current_wal_lsn();pg_current_wal_lsn 
--------------------0/1696F20
(1 row)postgres@slpc:~/pgsql$ pg_waldump -p pgdata/pg_wal/ -s 0/1696E28 -e 0/1696F20
rmgr: Heap        len (rec/tot):     54/   150, tx:        736, lsn: 0/01696E28, prev 0/01696DF0, desc: INSERT off 2 flags 0x00, blkref #0: rel 1663/13010/16384 blk 0 FPW
rmgr: Transaction len (rec/tot):     34/    34, tx:        736, lsn: 0/01696EC0, prev 0/01696E28, desc: COMMIT 2023-09-23 11:27:25.399199 CST
rmgr: Standby     len (rec/tot):     50/    50, tx:          0, lsn: 0/01696EE8, prev 0/01696EC0, desc: RUNNING_XACTS nextXid 737 latestCompletedXid 736 oldestRunningXid 737

pg_waldump源码分析

pg_waldump的源码比较简单，在src/bin/pg_waldump中，执行pg_waldump -p pgdata/pg_wal/ -s 0/1696E28 -e 0/1696F20进行调试。

main(int argc, char **argv)
--> XLogReaderAllocate(WalSegSz, waldir,XL_ROUTINE(.page_read = WALDumpReadPage,.segment_open = WALDumpOpenSegment, .segment_close = WALDumpCloseSegment), &private);// 因为我们是读wal日志的一个范围，从起点到终点，所以，这里第1步要做的就是先找到起始的wal日志记录
--> XLogFindNextRecord(xlogreader_state, private.startptr);for (;;){// 开始读取后续WAL日志/* try to read the next record */record = XLogReadRecord(xlogreader_state, &errormsg);/* perform any per-record work */if (!config.quiet){if (config.stats == true)XLogDumpCountRecord(&config, &stats, xlogreader_state);elseXLogDumpDisplayRecord(&config, xlogreader_state);}/* check whether we printed enough */config.already_displayed_records++;if (config.stop_after_records > 0 &&config.already_displayed_records >= config.stop_after_records)break;}
--> XLogReaderFree(xlogreader_state);

更详细的代码可以查看：

int main(int argc, char **argv)
{uint32		xlogid;uint32		xrecoff;XLogReaderState *xlogreader_state;XLogDumpPrivate private;XLogDumpConfig config;XLogDumpStats stats;XLogRecord *record;XLogRecPtr	first_record;private.timeline = 1;private.startptr = InvalidXLogRecPtr;     // 起点private.endptr = InvalidXLogRecPtr;       // 终点private.endptr_reached = false;while ((option = getopt_long(argc, argv, "be:fn:p:qr:s:t:x:z", long_options, &optindex)) != -1){switch (option){case 'e':if (sscanf(optarg, "%X/%X", &xlogid, &xrecoff) != 2){pg_log_error("could not parse end WAL location \"%s\"",optarg);goto bad_argument;}private.endptr = (uint64) xlogid << 32 | xrecoff;break;case 's':if (sscanf(optarg, "%X/%X", &xlogid, &xrecoff) != 2){pg_log_error("could not parse start WAL location \"%s\"",optarg);goto bad_argument;}elseprivate.startptr = (uint64) xlogid << 32 | xrecoff;break;// ...}}// .../* done with argument parsing, do the actual work *//* we have everything we need, start reading */// 初始化XLogReader，分配读取所需的内存空间xlogreader_state = XLogReaderAllocate(WalSegSz, waldir, XL_ROUTINE(.page_read = WALDumpReadPage,.segment_open = WALDumpOpenSegment,.segment_close = WALDumpCloseSegment), &private);/* first find a valid recptr to start from */// 找到第一个lsn >= startptr的有效日志记录first_record = XLogFindNextRecord(xlogreader_state, private.startptr);if (first_record == InvalidXLogRecPtr)fatal_error("could not find a valid record after %X/%X",LSN_FORMAT_ARGS(private.startptr));/** Display a message that we're skipping data if `from` wasn't a pointer* to the start of a record and also wasn't a pointer to the beginning of* a segment (e.g. we were used in file mode).*/if (first_record != private.startptr &&XLogSegmentOffset(private.startptr, WalSegSz) != 0)printf(ngettext("first record is after %X/%X, at %X/%X, skipping over %u byte\n","first record is after %X/%X, at %X/%X, skipping over %u bytes\n",(first_record - private.startptr)),LSN_FORMAT_ARGS(private.startptr),LSN_FORMAT_ARGS(first_record),(uint32) (first_record - private.startptr));for (;;){/* try to read the next record */record = XLogReadRecord(xlogreader_state, &errormsg);if (!record){if (!config.follow || private.endptr_reached)break;else{pg_usleep(1000000L);	/* 1 second */continue;}}/* apply all specified filters */if (config.filter_by_rmgr != -1 &&config.filter_by_rmgr != record->xl_rmid)continue;if (config.filter_by_xid_enabled &&config.filter_by_xid != record->xl_xid)continue;/* perform any per-record work */if (!config.quiet){if (config.stats == true)XLogDumpCountRecord(&config, &stats, xlogreader_state);elseXLogDumpDisplayRecord(&config, xlogreader_state);}/* check whether we printed enough */config.already_displayed_records++;if (config.stop_after_records > 0 &&config.already_displayed_records >= config.stop_after_records)break;}if (config.stats == true && !config.quiet)XLogDumpDisplayStats(&config, &stats);if (errormsg)fatal_error("error in WAL record at %X/%X: %s",LSN_FORMAT_ARGS(xlogreader_state->ReadRecPtr),errormsg);XLogReaderFree(xlogreader_state);return EXIT_SUCCESS;bad_argument:fprintf(stderr, _("Try \"%s --help\" for more information.\n"), progname);return EXIT_FAILURE;
}