浅谈PostgreSQL中的并发实现

typedef  struct  HeapTupleFields 
 { 
 	//数据时候的事务ID
  	TransactionId t_xmin ; 		
	// 数据删除时间的事务id
  	TransactionId t_xmax ; 		
 
 	union 
 	{ 
		// 插入或删除命令id
  		CommandId t_cid ; 	
		// 表格最做fullvacuum的事务ID
  		TransactionId t_xvac ; 	
 	} 			t_field3 ; 
 } HeapTupleFields ; 
 
 typedef  struct  DatumTupleFields 
 { 
 	// 修改首部的长度
 	int32 datum_len_ ; 		
	// -1 或者记录的唯一标识
 	int32 datum_typmod ; 	/* -1，或记录类型的标识符 */ 
 	// 符合类型的oid或记录id
 	Oid datum_typeid ; 	/* 复合类型 OID 或 RECORDOID */ 

 } DatumTupleFields ; 
 
 结构 HeapTupleHeaderData 
 { 
 	union 
 	{
  		HeapTupleFields t_heap ; 
 		DatumTupleFields t_datum ; 
 	} 			t_choice ; 
 	
	// 元组未删除，保存当前记录的位置；如果删除，保存新记录的位置
 	ItemPointerData t_ctid ; 		
 
 	// 属性和标记位
 # define  FIELDNO_HEAPTUPLEHEADERDATA_INFOMASK2  2
  	uint16 t_infomask2 ; 	
 	// 标记位
 #define FIELDNO_HEAPTUPLEHEADERDATA_INFOMASK  3 
  	uint16 t_infomask ;		
 	// header的位图填充
 #define FIELDNO_HEAPTUPLEHEADERDATA_HOFF  4 
  	uint8 t_hoff ; 			
	// NULL的位图填充
 #define FIELDNO_HEAPTUPLEHEADERDATA_BITS  5 
  	bits8 t_bits [ FLEXIBLE_ARRAY_MEMBER ] ; 	/* NULL 位图 */ 
 
 } ;

$ /usr/local/postgres/bin/psql  -h 127.0.0.1 -d postgres
postgres=#  create extension pageinspect;
CREATE EXTENSION
postgres=# create table test1(a1 int,a2 int);
CREATE TABLE
postgres=# begin;
BEGIN
postgres=*# insert into test1 values(1,1);
INSERT 0 1
postgres=*# select txid_current();
 txid_current 
--------------
          740
(1 row)

postgres=*# \d
          List of relations
 Schema | Name  | Type  |   Owner    
--------+-------+-------+------------
 public | test1 | table | perrynzhou
(1 row)

postgres=*# select t_xmin, t_xmax,t_field3,t_infomask, t_infomask2,t_data  from heap_page_items(get_raw_page('public.test1', 0));
 t_xmin | t_xmax | t_field3 | t_infomask | t_infomask2 |       t_data       
--------+--------+----------+------------+-------------+--------------------
    740 |      0 |        0 |       2048 |           2 | \x0100000001000000
(1 row)

postgres=*# commit;
COMMIT
postgres=# select t_xmin, t_xmax,t_field3,t_infomask, t_infomask2,t_data  from heap_page_items(get_raw_page('public.test1', 0));
 t_xmin | t_xmax | t_field3 | t_infomask | t_infomask2 |       t_data       
--------+--------+----------+------------+-------------+--------------------
    740 |      0 |        0 |       2048 |           2 | \x0100000001000000
(1 row)

// 执行一次查询发现t_infomask 由2048更新为了2304
postgres=# select * from test1;
 a1 | a2 
----+----
  1 |  1
(1 row)

postgres=# select t_xmin, t_xmax,t_field3,t_infomask, t_infomask2,t_data  from heap_page_items(get_raw_page('public.test1', 0));
 t_xmin | t_xmax | t_field3 | t_infomask | t_infomask2 |       t_data       
--------+--------+----------+------------+-------------+--------------------
    740 |      0 |        0 |       2304 |           2 | \x0100000001000000
(1 row)

postgres=#

// 快照类型的定义
typedef enum SnapshotType
{
	
	// 本事务插入，元组可见;符合事务快照元组可见
	SNAPSHOT_MVCC = 0,

	// 事务提交或者当前事务运行中，则元组是可见
	SNAPSHOT_SELF,

	// 没有任何限制，元组都可见
	SNAPSHOT_ANY,

	// 用于检查toast表的可见性
	SNAPSHOT_TOAST,

	// 事务提交或者终止，可见性和SNAPSHOT_SELF保持一致；如果是进行中的写入事务，则它的可见性和SNAPSHOT_SELF不一致，此时会收集元组版本信息保存到快照中
	SNAPSHOT_DIRTY,
	
	// 用于逻辑复制中逻辑编解码的可见性判断
	SNAPSHOT_HISTORIC_MVCC,

	//  判断元组是否对某些事务可见
	SNAPSHOT_NON_VACUUMABLE
} SnapshotType;

// PG中事务的快照的定义
typedef struct SnapshotData
{
	// 事务快照类型
	SnapshotType snapshot_type; 

	// 当前活跃事务中最小的事务ID
	TransactionId xmin;			
	// 当前活跃事务中已提交的最大事务ID
	TransactionId xmax;			

	// xmin和xmax之间的事务
	TransactionId *xip;
	uint32		xcnt;			/* # of xact ids in xip[] */

	// 子事务列表
	TransactionId *subxip;
	int32		subxcnt;		/* # of xact ids in subxip[] */
	bool		suboverflowed;	/* has the subxip array overflowed? */

	bool		takenDuringRecovery;	/* recovery-shaped snapshot? */
	bool		copied;			/* false if it's a static snapshot */

	CommandId	curcid;			/* in my xact, CID < curcid are visible */

	/*
	 * An extra return value for HeapTupleSatisfiesDirty, not used in MVCC
	 * snapshots.
	 */
	uint32		speculativeToken;

	/*
	 * For SNAPSHOT_NON_VACUUMABLE (and hopefully more in the future) this is
	 * used to determine whether row could be vacuumed.
	 */
	struct GlobalVisState *vistest;

	/*
	 * Book-keeping information, used by the snapshot manager
	 */
	uint32		active_count;	/* refcount on ActiveSnapshot stack */
	uint32		regd_count;		/* refcount on RegisteredSnapshots */
	pairingheap_node ph_node;	/* link in the RegisteredSnapshots heap */

	// 快照的时间戳
	TimestampTz whenTaken;		/* timestamp when snapshot was taken */
	// redo的LSN
	XLogRecPtr	lsn;			/* position in the WAL stream when taken */

	/*
	 * The transaction completion count at the time GetSnapshotData() built
	 * this snapshot. Allows to avoid re-computing static snapshots when no
	 * transactions completed since the last GetSnapshotData().
	 */
	uint64		snapXactCompletionCount;
} SnapshotData;