PostgreSQL源码分析——pg_basebackup

涉及到的代码主要在src/backend/replication以及bin/pg_basebackup中。

我们知道pg_basebackup是一个进行基础备份的工具，除了使用这个工具，还可以用底层API的方式进行基础备份，主要过程如下：

1. 连接到数据库
2. 执行select pg_start_backup('lable')命令。（会强制发生一次checkpoint，并将检查点记录到backup_label文件中）
3. 执行备份，把数据目录进行复制（包含backup_label）
4. 执行select pg_stop_backup命令，（删除backup_label文件，并在WAL日志中写入一条XLOG_BACKUP_END的记录，当备节点回放到该记录时，就知道备份结束了，数据达到了一致点，可以对外提供服务了）
5. 备份过程中产生的WAL日志进行复制

其实，pg_basebackup工具就是对底层API的封装，其主要过程是相同的，但具体到代码，并不是直接调用的pg_start_backup,pg_stop_backup函数，而是通过一些命令的形式，这些特殊的命令定义在src/backend/replication/repl_gram.y中，后面我们会进行分析。

主流程

pg_basebackup执行基础备份的主要流程如下，其中，涉及到libpq协议与服务端进行连接，通信，向服务端发送一些特殊的命令语句，这些命令的解析在src/backend/replication/repl_gram.y中可以查看到具体的语法定义。主流程如下：

main(int argc, char **argv)
--> GetConnection();   // 连接服务端，（例如主节点）--> PQconnectdbParams(keywords, values, true);
--> BaseBackup();      // 执行基础备份--> GenerateRecoveryConfig(conn, replication_slot);   // 用于生成primary_conninfo配置信息--> PQconninfo(pgconn);--> RunIdentifySystem(conn, &sysidentifier, &latesttli, NULL, NULL)basebkp = psprintf("BASE_BACKUP LABEL '%s' %s %s %s %s %s %s %s %s %s",escaped_label,estimatesize ? "PROGRESS" : "",includewal == FETCH_WAL ? "WAL" : "",fastcheckpoint ? "FAST" : "",includewal == NO_WAL ? "" : "NOWAIT",maxrate_clause ? maxrate_clause : "",format == 't' ? "TABLESPACE_MAP" : "",verify_checksums ? "" : "NOVERIFY_CHECKSUMS",manifest_clause ? manifest_clause : "",manifest_checksums_clause);// 向服务端发送执行命令"BASE_BACKUP LABEL 'pg14bak' PROGRESS   NOWAIT    MANIFEST 'yes' "--> PQsendQuery(conn, basebkp)    --> PQgetResult(conn);    // Get the starting WAL location--> StartLogStreamer(xlogstart, starttli, sysidentifier);--> CreateReplicationSlot(param->bgconn, replication_slot, NULL, temp_replication_slot, true, true, false)// 执行命令 "CREATE_REPLICATION_SLOT \"pg_basebackup_2553309\" TEMPORARY PHYSICAL RESERVE_WAL"--> PQexec(conn, query->data);// Start a child process and tell it to start streaming.// 创建一个单独的子进程用于日志传输bgchild = fork();if (bgchild == 0){/* in child process */LogStreamerMain(param);--> ReceiveXlogStream(param->bgconn, &stream)   // Receive a log stream starting at the specified position.}if (!writing_to_stdout && manifest)ReceiveBackupManifest(conn);    // receive backup manifest

整个的过程，最重要的有3点：

1. 在进行备份前，执行一次checkpoint，记录开始的位置，在服务端接收到BASE_BACKUP LABLE命令后，生成备份标签文件backup_lable，这个文件最重要的作用是记录数据库恢复的起始位置。当启动备份实例时，会读该文件进行恢复。
2. 复制数据库数据文件
3. 日志复制

我们可以看一下backup_lable文件中的内容：

postgres@slpc:~/pgsql/pgbak$ ls
backup_label      pg_commit_ts   pg_notify     pg_subtrans  postgresql.auto.conf
backup_manifest   pg_dynshmem    pg_replslot   pg_tblspc    postgresql.conf
base              pg_hba.conf    pg_serial     pg_twophase  standby.signal
current_logfiles  pg_ident.conf  pg_snapshots  PG_VERSION
global            pg_logical     pg_stat       pg_wal
log               pg_multixact   pg_stat_tmp   pg_xact
postgres@slpc:~/pgsql/pgbak$ cat backup_label 
START WAL LOCATION: 0/C000028 (file 00000001000000000000000C)    备份开始时日志的位置
CHECKPOINT LOCATION: 0/C000060                                   检查点的位置
BACKUP METHOD: streamed                                          备份方法
BACKUP FROM: primary                                             备份源    
START TIME: 2023-08-02 17:15:58 CST                              备份开始的物理时间
LABEL: pg14bak                                                   备份标签
START TIMELINE: 1

具体过程

仅看上面的主流程还是有一些不清楚的地方的。

这里有个很重要的命令BASE_BACKUP LABEL，备份命令，获取XLOG的存放路径和备份开始时日志的位置，那么服务端这块是怎么处理的呢？我们看一下服务端的相关代码：

void PostgresMain(int argc, char *argv[], const char *dbname, const char *username)
{// ...if (am_walsender)WalSndSignals();/* Perform initialization specific to a WAL sender process. */if (am_walsender)InitWalSender();for (;;){firstchar = ReadCommand(&input_message);switch (firstchar){case 'Q':			/* simple query */{const char *query_string;/* Set statement_timestamp() */SetCurrentStatementStartTimestamp();query_string = pq_getmsgstring(&input_message);pq_getmsgend(&input_message);if (am_walsender){if (!exec_replication_command(query_string))exec_simple_query(query_string);}elseexec_simple_query(query_string);send_ready_for_query = true;}break;}
}/* Execute an incoming replication command.*/
bool exec_replication_command(const char *cmd_string)
{// .../* Looks like a WalSender command, so parse it. */parse_rc = replication_yyparse();if (parse_rc != 0)ereport(ERROR,(errcode(ERRCODE_SYNTAX_ERROR),errmsg_internal("replication command parser returned %d",parse_rc)));switch (cmd_node->type){case T_IdentifySystemCmd:cmdtag = "IDENTIFY_SYSTEM";set_ps_display(cmdtag);IdentifySystem();EndReplicationCommand(cmdtag);break;case T_BaseBackupCmd:cmdtag = "BASE_BACKUP";set_ps_display(cmdtag);PreventInTransactionBlock(true, cmdtag);SendBaseBackup((BaseBackupCmd *) cmd_node);EndReplicationCommand(cmdtag);break;case T_CreateReplicationSlotCmd:cmdtag = "CREATE_REPLICATION_SLOT";set_ps_display(cmdtag);CreateReplicationSlot((CreateReplicationSlotCmd *) cmd_node);EndReplicationCommand(cmdtag);break;case T_DropReplicationSlotCmd:cmdtag = "DROP_REPLICATION_SLOT";set_ps_display(cmdtag);DropReplicationSlot((DropReplicationSlotCmd *) cmd_node);EndReplicationCommand(cmdtag);break;case T_StartReplicationCmd:{StartReplicationCmd *cmd = (StartReplicationCmd *) cmd_node;cmdtag = "START_REPLICATION";set_ps_display(cmdtag);PreventInTransactionBlock(true, cmdtag);if (cmd->kind == REPLICATION_KIND_PHYSICAL)StartReplication(cmd);elseStartLogicalReplication(cmd);/* dupe, but necessary per libpqrcv_endstreaming */EndReplicationCommand(cmdtag);Assert(xlogreader != NULL);break;}case T_TimeLineHistoryCmd:cmdtag = "TIMELINE_HISTORY";set_ps_display(cmdtag);PreventInTransactionBlock(true, cmdtag);SendTimeLineHistory((TimeLineHistoryCmd *) cmd_node);EndReplicationCommand(cmdtag);break;case T_VariableShowStmt:{DestReceiver *dest = CreateDestReceiver(DestRemoteSimple);VariableShowStmt *n = (VariableShowStmt *) cmd_node;cmdtag = "SHOW";set_ps_display(cmdtag);/* syscache access needs a transaction environment */StartTransactionCommand();GetPGVariable(n->name, dest);CommitTransactionCommand();EndReplicationCommand(cmdtag);}break;default:elog(ERROR, "unrecognized replication command node tag: %u",cmd_node->type);}// ...
}

我们继续看一下SendBaseBackup函数的实现。

/** SendBaseBackup() - send a complete base backup.** The function will put the system into backup mode like pg_start_backup()* does, so that the backup is consistent even though we read directly from* the filesystem, bypassing the buffer cache.*/
void SendBaseBackup(BaseBackupCmd *cmd)
{basebackup_options opt;SessionBackupState status = get_backup_status();if (status == SESSION_BACKUP_NON_EXCLUSIVE)ereport(ERROR,(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),errmsg("a backup is already in progress in this session")));parse_basebackup_options(cmd->options, &opt);WalSndSetState(WALSNDSTATE_BACKUP);if (update_process_title){char		activitymsg[50];snprintf(activitymsg, sizeof(activitymsg), "sending backup \"%s\"",opt.label);set_ps_display(activitymsg);}perform_base_backup(&opt);
}

主流程如下：

SendBaseBackup(BaseBackupCmd *cmd)
--> perform_base_backup(&opt);// creates the necessary starting checkpoint and constructs the backup label file.--> do_pg_start_backup(opt->label, opt->fastcheckpoint, &starttli, labelfile, &tablespaces, tblspc_map_file);--> RequestCheckpoint(CHECKPOINT_FORCE | CHECKPOINT_WAIT | (fast ? CHECKPOINT_IMMEDIATE : 0));--> CreateCheckPoint(flags | CHECKPOINT_IMMEDIATE);