PostgreSQL技术内幕9：PostgreSQL事务原理解析

文章目录

- 0.简介
- 1.PG事务整体介绍
- - 1.1 事务类型介绍
- 1.2 事务模块介绍
- 2. 代码分析

0.简介

有了上一篇数据库事务并发控制协议的介绍，对于数据库事务和并发控制有了基本的认识，本文将介绍PG事务模块，主要介绍PG支持的事务类型（普通事务，子事务，多事务，2pc事务），事务模块涉及的的子模块整体介绍（日志模块，MVCC管理，事务锁管理）和事务模块代码脉络解读。日志模块、MVCC管理和事务锁管理详细内容将在后续文章中介绍。

1.PG事务整体介绍

1.1 事务类型介绍

从事务类型上来区分，PG包含普通的事务，子事务，多事务，2PC事务。
普通事务：
普通事务是数据库操作的基本单位，用于确保数据的完整性和一致性。在PostgreSQL中，普通事务可以通过BEGIN命令开始，通过COMMIT命令提交，或通过ROLLBACK命令回滚。在默认情况下，如果客户端没有显式地开始一个事务，PostgreSQL会在执行每个DML（数据操纵语言）语句时自动提交事务，即每条DML语句都被视为一个单独的事务。但用户可以通过设置AUTOCOMMIT为off或使用BEGIN命令来显式地控制事务的开始和结束。
用PG客户端直接连接PG服务器会默认开始事务的自动提交，就是把每一句SQL当成一个事务自动提交。可以使用set autocommit off关闭自动提交，或者使用begin开启一个事务块（一组被BEGIN和COMMIT包围的语句也被称为一个事务块。），下面用一个例子来具体看一看：

postgres=# \echo :AUTOCOMMIT
on
postgres=# create table t3(a int);
CREATE TABLE
postgres=# \set AUTOCOMMIT off
postgres=# \echo :AUTOCOMMIT
off
postgres=# insert into t3 values(1);
INSERT 0 1
postgres=# rollback;
ROLLBACK
postgres=# select * from t3;
 a
---
(0 rows)

postgres=# insert into t3 values(2);
INSERT 0 1
postgres=# commit;
COMMIT
postgres=# select * from t3;
 a
---
 2
(1 row)

可以使用如下语句查看数据信息，其中记录所有事务的状态：

select lp,t_xmin,t_xmax,t_ctid,t_infomask,t_data from heap_page_items(get_raw_page('t3',0));

事务信息物理存储在PGDATA目录下的pg_xact文件夹，这里的文件记录了事务的提交状态。2bit记录一个事务的状态，一个byte可以记录4个事务状态，这里每一个文件的size在内核中规定为32BLCKSZ,比如0000文件所记录的事务ID范围是(1~32BLCKSZ*4)。因此想要获取一个事务的提交状态，到pg_xact目录下取就可以了。
子事务
子事务是在一个普通事务内部创建的更小的事务单位。它们允许部分事务的回滚，而不需要回滚整个事务。在PostgreSQL中，子事务通过SAVEPOINT命令来创建保存点，并通过ROLLBACK TO SAVEPOINT命令回滚到指定的保存点。虽然子事务不能直接提交，但它们会随着父事务的提交而提交。子事务在处理大量数据或复杂逻辑时非常有用，因为它允许在出现问题时仅回滚部分操作，而不是整个事务。
子事务是调用savepoint或者exception而出现的，多个子事务共同组成一个完整的事务。

postgres=# begin;
BEGIN
postgres=# insert into t3 values(6);
INSERT 0 1
postgres=# savepoint s1;
SAVEPOINT
postgres=# insert into t3 values(7);
INSERT 0 1
postgres=# savepoint s2;
SAVEPOINT
postgres=# select txid_current();
 txid_current
--------------
          585
(1 row)

postgres=# commit;
COMMIT
postgres=# select ctid,xmin,* from t3;
 ctid  | xmin | a
-------+------+---
 (0,2) |  581 | 2
 (0,4) |  585 | 6
 (0,5) |  586 | 7
(3 rows)

可以看到，数据有着不同的事务id，585为父事务，586为子事务，子事务可以当成是一种特殊的普通事务，也有提及状态吗，也会再pg_xact中记录，所有的父事务和子事务，最用都要标记为commit或者abort，为了保证整个父事务和子事务的原则性，提交机制如下：
1）子事务先标记为子事务提交
(TRANSACTION_STATUS_SUB_COMMITTED)
2)标记父事务状态为已提交
(TRANSACTION_STATUS_COMMITTED)
3）标记子事务状态为已提交
(TRANSACTION_STATUS_COMMITTED)
如果子事务在pg_xact目录记录状态为已提交，那么子事务就是提交状态；如果子事务状态为子事务提交，子事务状态的判断需要根据父事务的状态来判断；如果是其他情况，就是未提交。子事务和父事务的对应关系记录在pg_subtrans目录下。
在文件中，使用4byte记录一个子事务父事务id，每个文件size在内核中规定为32blcksz，比如0000文件记录的事务id范围为（1~32blcksz/4)
多事务
多事务通常与行级锁相关，出现在多个会话对同一行记录添加行级锁时。在这种情况下，每个会话都可以视为一个独立的事务，但它们的操作可能会相互影响，因为数据库需要管理这些事务对共享资源的访问。多事务的处理是数据库并发控制的一部分，PostgreSQL通过其MVCC（多版本并发控制）机制来管理这种并发情况，确保数据的一致性和隔离性。
多事务记录在pg_multixact目录下，里面记录了多事务和其对应事务列表的关联关系，事务列表中事务属性决定了多事务的提交状态。
2PC事务
2PC事务是一种分布式事务处理协议，用于确保在多个数据库或事务性资源之间执行的事务的原子性。2PC事务分为两个阶段：准备阶段和提交阶段。在准备阶段，所有参与者（如数据库）都会准备提交事务，但在实际提交之前，它们会等待协调者的进一步指示。如果所有参与者都准备好了，协调者会发出提交命令，所有参与者将同时提交事务。如果任何参与者在准备阶段失败，协调者将发出回滚命令，所有参与者将回滚事务。
PG中通过prepare和commit prepared xxx语句来执行，比如：prepare transaction '2pc_trans’是第一阶段提交，commit prepared ‘2pc_trans’;是第二阶段提交。2PC事务可以认为是一种特殊的普通事务，其提交状态判断和正常事务一样，但是2pc事务可以不依赖Session连接，数据库关闭也不会影响2PC事务状态，其保存的事务数据文件在pg_twophase中。

1.2 事务模块介绍

从模块架构来看，分为日志模块，MVCC管理模块和锁管理模块

在这里插入图片描述

2. 代码分析

代码整体结构可以参考src/backend/access/transam/README，以下是部分内容和翻译：

PostgreSQL's transaction system is a three-layer system.  The bottom layer
implements low-level transactions and subtransactions, on top of which rests
the mainloop's control code, which in turn implements user-visible
transactions and savepoints.
PG事务分为3层,底层实现了低层次的事务和子事务,在其顶上驻留主循环控制代码,
而主循环实现了用户可见性事务和保存点.

The middle layer of code is called by postgres.c before and after the
processing of each query, or after detecting an error:
代码的中间层由 postgres.c 调用在处理每个查询流程，或者在检测到错误后：
    StartTransactionCommand
    CommitTransactionCommand
    AbortCurrentTransaction

Meanwhile, the user can alter the system's state by issuing the SQL commands
BEGIN, COMMIT, ROLLBACK, SAVEPOINT, ROLLBACK TO or RELEASE.  The traffic cop
redirects these calls to the toplevel routines
同时，用户可以通过发出 SQL 命令来更改系统的状态BEGIN、COMMIT、ROLLBACK、SAVEPOINT、
ROLLBACK TO 或 RELEASE。 将这些调用重定向到顶级例程
    BeginTransactionBlock
    EndTransactionBlock
    UserAbortTransactionBlock
    DefineSavepoint
    RollbackToSavepoint
    ReleaseSavepoint

respectively.  Depending on the current state of the system, these functions
call low level functions to activate the real transaction system:
分别。 根据系统的当前状态，分别调用这些功能 low level 函数来激活 
Real Transaction 系统：
    StartTransaction
    CommitTransaction
    AbortTransaction
    CleanupTransaction
    StartSubTransaction
    CommitSubTransaction
    AbortSubTransaction
    CleanupSubTransaction

For example, consider the following sequence of user commands:

1)    BEGIN
2)    SELECT * FROM foo
3)    INSERT INTO foo VALUES (...)
4)    COMMIT

In the main processing loop, this results in the following function call
sequence:

     /  StartTransactionCommand;
    /       StartTransaction;
1) <    ProcessUtility;                 << BEGIN
    \       BeginTransactionBlock;
     \  CommitTransactionCommand;

    /   StartTransactionCommand;
2) /    PortalRunSelect;                << SELECT ...
   \    CommitTransactionCommand;
    \       CommandCounterIncrement;

    /   StartTransactionCommand;
3) /    ProcessQuery;                   << INSERT ...
   \    CommitTransactionCommand;
    \       CommandCounterIncrement;

     /  StartTransactionCommand;
    /   ProcessUtility;                 << COMMIT
4) <        EndTransactionBlock;
    \   CommitTransactionCommand;
     \      CommitTransaction;


事务状态结构

typedef enum TransState
{
  TRANS_DEFAULT,        /* idle */
  TRANS_START,        /* transaction starting */
  TRANS_INPROGRESS,      /* inside a valid transaction */
  TRANS_COMMIT,        /* commit in progress */
  TRANS_ABORT,        /* abort in progress */
  TRANS_PREPARE,        /* prepare in progress */
} TransState;

/*
 *  transaction block states - transaction state of client queries
 *
 * Note: the subtransaction states are used only for non-topmost
 * transactions; the others appear only in the topmost transaction.
 */
typedef enum TBlockState
{
  /* not-in-transaction-block states */
  TBLOCK_DEFAULT,        /* idle */
  TBLOCK_STARTED,        /* running single-query transaction */

  /* transaction block states */
  TBLOCK_BEGIN,        /* starting transaction block */
  TBLOCK_INPROGRESS,      /* live transaction */
  TBLOCK_IMPLICIT_INPROGRESS, /* live transaction after implicit BEGIN */
  TBLOCK_PARALLEL_INPROGRESS, /* live transaction inside parallel worker */
  TBLOCK_END,          /* COMMIT received */
  TBLOCK_ABORT,        /* failed xact, awaiting ROLLBACK */
  TBLOCK_ABORT_END,      /* failed xact, ROLLBACK received */
  TBLOCK_ABORT_PENDING,    /* live xact, ROLLBACK received */
  TBLOCK_PREPARE,        /* live xact, PREPARE received */

  /* subtransaction states */
  TBLOCK_SUBBEGIN,      /* starting a subtransaction */
  TBLOCK_SUBINPROGRESS,    /* live subtransaction */
  TBLOCK_SUBRELEASE,      /* RELEASE received */
  TBLOCK_SUBCOMMIT,      /* COMMIT received while TBLOCK_SUBINPROGRESS */
  TBLOCK_SUBABORT,      /* failed subxact, awaiting ROLLBACK */
  TBLOCK_SUBABORT_END,    /* failed subxact, ROLLBACK received */
  TBLOCK_SUBABORT_PENDING,  /* live subxact, ROLLBACK received */
  TBLOCK_SUBRESTART,      /* live subxact, ROLLBACK TO received */
  TBLOCK_SUBABORT_RESTART,  /* failed subxact, ROLLBACK TO received */
} TBlockState;

/*
 *  transaction state structure
 */
typedef struct TransactionStateData
{
  FullTransactionId fullTransactionId;  /* my FullTransactionId */
  SubTransactionId subTransactionId;  /* my subxact ID */
  char     *name;      /* savepoint name, if any */
  int      savepointLevel; /* savepoint level */
  TransState  state;      /* low-level state */
  TBlockState blockState;    /* high-level state */
  int      nestingLevel;  /* transaction nesting depth */
  int      gucNestLevel;  /* GUC context nesting depth */
  MemoryContext curTransactionContext;  /* my xact-lifetime context */
  ResourceOwner curTransactionOwner;  /* my query resources */
  TransactionId *childXids;  /* subcommitted child XIDs, in XID order */
  int      nChildXids;    /* # of subcommitted child XIDs */
  int      maxChildXids;  /* allocated size of childXids[] */
  Oid      prevUser;    /* previous CurrentUserId setting */
  int      prevSecContext; /* previous SecurityRestrictionContext */
  bool    prevXactReadOnly;  /* entry-time xact r/o state */
  bool    startedInRecovery;  /* did we start in recovery? */
  bool    didLogXid;    /* has xid been included in WAL record? */
  int      parallelModeLevel;  /* Enter/ExitParallelMode counter */
  bool    chain;      /* start a new block after this one */
  bool    topXidLogged;  /* for a subxact: is top-level XID logged? */
  struct TransactionStateData *parent;  /* back link to parent */
} TransactionStateData;