Postgres逻辑解码中HeapTuple的行为

零、前言

本文主要分析Postgres 9.4的代码，以及Postgre 12.12和GPDB 7的代码，后两者代码相同。目前GPDB7尚不支持逻辑解码。

一、为了支持逻辑解码Postgres对WAL做了哪些修改

Postgres 9.4引入逻辑解码的功能。对XLog修改的commit link: Add new wal_level, logical, sufficient for logical decoding.

简单分析其中的代码，代码来自Postgres 9.4：

RelationPutHeapTuple(relation, buffer, heaptup)              // 把HeapTuple信息放入Buffer 
bool		need_tuple_data;                             // 是否需要HeapTuple
need_tuple_data = RelationIsLogicallyLogged(relation);       // 逻辑日志则需要HeapTuple
rdata[1].buffer = need_tuple_data ? InvalidBuffer : buffer;  // 逻辑日志则给rdata[1]赋值为HeapTuple，物理日志为空

可以得出结论，逻辑日志需要把HeapTuple写入WAL，但是物理日志不需要。这是为了支持逻辑解码对WAL做的最大改动。事实上，这个commit以前没有逻辑日志，但是也把HeapTuple写到物理日志里去了，具体原因目前不知，但也因此过渡的十分平滑。

二、HeapTuple信息写入WAL的流程

为了更方便和GPDB7做对比，以下代码来自Postgres 12.12的heap_insert()函数。
XLogRegisterData()用来向XLOG写入main_data。
XLogRegisterBufData()用来想XLOG写入HeapTuple（可能还有其它功能）。

void
heap_insert(Relation relation, HeapTuple tup, CommandId cid,
			int options, BulkInsertState bistate, TransactionId xid)
{
	HeapTuple	heaptup;
	heaptup = heap_prepare_insert(relation, tup, xid, cid, options, isFrozen);

	if (needwal) // 是否需要写WAL
	{
		xl_heap_insert xlrec;
		xl_heap_header xlhdr;
		XLogRecPtr	recptr;
		Page		page = BufferGetPage(buffer);
		uint8		info = XLOG_HEAP_INSERT;
        	// 写入日志前的准备
		XLogBeginInsert();
		// 注册生成日志相关的数据
		// XLogRegisterData()用来向XLOG写入main_data。
		XLogRegisterData((char *) &xlrec, SizeOfHeapInsert);

		XLogRegisterBuffer(0, buffer, REGBUF_STANDARD | bufflags);
		
		XLogRegisterBufData(0, (char *) &xlhdr, SizeOfHeapHeader);
		// XLogRegisterBufData()用来想XLOG写入HeapTuple。
		XLogRegisterBufData(0,
							(char *) heaptup->t_data + SizeofHeapTupleHeader,
							heaptup->t_len - SizeofHeapTupleHeader);
		// 执行日志插入，写入到缓冲区，目前尚未落盘
		recptr = XLogInsert(RM_HEAP_ID, info);
		// 设置page的LSN
		PageSetLSN(page, recptr);
	}

}

三、HeapTuple信息被逻辑解码的流程

XLogReaderState是用于读取XLog信息的结构体，以下只列出两个主要变量。main_data存储XLog的基本信息，HeapTuple信息会被存储在blocks中。

struct XLogReaderState
{
	char	   *main_data;		/* record's main data portion */
	/* information about blocks referenced by the record. */
	DecodedBkpBlock blocks[XLR_MAX_BLOCK_ID + 1];
};

XLogRecGetData()可以获取XLogReaderState中的main_data，与上述XLogRegisterData()是一对。
XLogRecGetBlockData()可以获取XLogReaderState中的blocks，与上述XLogRegisterBufData()是一对。

static void
DecodeInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
{
	Size		datalen;
	char	   *tupledata;
	Size		tuplelen;
	XLogReaderState *r = buf->record;
	xl_heap_insert *xlrec;
	// XLogRecGetData()可以获取XLogReaderState中的main_data。
 	// 也就是物理日志里需要的协助准确定位的信息
	xlrec = (xl_heap_insert *) XLogRecGetData(r);

	memcpy(&change->data.tp.relnode, &target_node, sizeof(RelFileNode));
	// XLogRecGetBlockData()可以获取XLogReaderState中的blocks。
	// 此处获取的信息事实上就是HeapTuple相关信息
	tupledata = XLogRecGetBlockData(r, 0, &datalen);
	tuplelen = datalen - SizeOfHeapHeader;

	change->data.tp.newtuple =
		ReorderBufferGetTupleBuf(ctx->reorder, tuplelen);
	// 解码相关信息，获取HeapTuple
	DecodeXLogTuple(tupledata, datalen, change->data.tp.newtuple);
}