src/backend/replication/logical/reorderbuffer.c
static void
ReorderBufferSerializeTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
{
	dlist_iter	subtxn_i;
	dlist_mutable_iter change_i;
	int			fd = -1;
	XLogSegNo	curOpenSegNo = 0;
	Size		spilled = 0;
	char		path[MAXPGPATH];

	elog(DEBUG2, "spill %u changes in XID %u to disk",
		 (uint32) txn->nentries_mem, txn->xid);

	/* do the same to all child TXs */
	dlist_foreach(subtxn_i, &txn->subtxns)
	{
		ReorderBufferTXN *subtxn;

		subtxn = dlist_container(ReorderBufferTXN, node, subtxn_i.cur);
		ReorderBufferSerializeTXN(rb, subtxn);
	}

	/* serialize changestream */
	dlist_foreach_modify(change_i, &txn->changes)
	{
		ReorderBufferChange *change;

		change = dlist_container(ReorderBufferChange, node, change_i.cur);

		/*
		 * store in segment in which it belongs by start lsn, don't split over
		 * multiple segments tho
		 */
		if (fd == -1 || !XLByteInSeg(change->lsn, curOpenSegNo))
		{
			XLogRecPtr	recptr;

			if (fd != -1)
				CloseTransientFile(fd);

			XLByteToSeg(change->lsn, curOpenSegNo);
			XLogSegNoOffsetToRecPtr(curOpenSegNo, 0, recptr);

			/*
			 * No need to care about TLIs here, only used during a single run,
			 * so each LSN only maps to a specific WAL record.
			 */
			sprintf(path, "pg_replslot/%s/xid-%u-lsn-%X-%X.snap",
					NameStr(MyReplicationSlot->data.name), txn->xid,
					(uint32) (recptr >> 32), (uint32) recptr);

			/* open segment, create it if necessary */
			fd = OpenTransientFile(path,
								   O_CREAT | O_WRONLY | O_APPEND | PG_BINARY,
								   S_IRUSR | S_IWUSR);

			if (fd < 0)
				ereport(ERROR,
						(errcode_for_file_access(),
						 errmsg("could not open file \"%s\": %m",
								path)));
		}

		ReorderBufferSerializeChange(rb, txn, fd, change);
		dlist_delete(&change->node);
		ReorderBufferReturnChange(rb, change);

		spilled++;
	}

	Assert(spilled == txn->nentries_mem);
	Assert(dlist_is_empty(&txn->changes));
	txn->nentries_mem = 0;

	if (fd != -1)
		CloseTransientFile(fd);
}
/*
 * Spill data of a large transaction (and its subtransactions) to disk.
 */
Spill data of a large transaction (and its subtransactions) to disk.
Spill data of a large transaction ( and its subtransactions ) to disk .
static void
ReorderBufferSerializeTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
{
	dlist_iter	subtxn_i;
	dlist_mutable_iter change_i;
	int			fd = -1;
	XLogSegNo	curOpenSegNo = 0;
	Size		spilled = 0;
	char		path[MAXPGPATH];
	elog(DEBUG2, "spill %u changes in XID %u to disk",
		 (uint32) txn->nentries_mem, txn->xid);
	dlist_foreach(subtxn_i, &txn->subtxns)
	{
		ReorderBufferTXN *subtxn;
		subtxn = dlist_container(ReorderBufferTXN, node, subtxn_i.cur);
		ReorderBufferSerializeTXN(rb, subtxn);
	}
	dlist_foreach_modify(change_i, &txn->changes)
	{
		ReorderBufferChange *change;
		change = dlist_container(ReorderBufferChange, node, change_i.cur);
		if (fd == -1 || !XLByteInSeg(change->lsn, curOpenSegNo))
		{
			XLogRecPtr	recptr;
			if (fd != -1)
				CloseTransientFile(fd);
			XLByteToSeg(change->lsn, curOpenSegNo);
			XLogSegNoOffsetToRecPtr(curOpenSegNo, 0, recptr);
			sprintf(path, "pg_replslot/%s/xid-%u-lsn-%X-%X.snap",
					NameStr(MyReplicationSlot->data.name), txn->xid,
					(uint32) (recptr >> 32), (uint32) recptr);
			fd = OpenTransientFile(path,
								   O_CREAT | O_WRONLY | O_APPEND | PG_BINARY,
								   S_IRUSR | S_IWUSR);
			if (fd < 0)
				ereport(ERROR,
						(errcode_for_file_access(),
						 errmsg("could not open file \"%s\": %m",
								path)));
		}
		ReorderBufferSerializeChange(rb, txn, fd, change);
		dlist_delete(&change->node);
		ReorderBufferReturnChange(rb, change);
		spilled++;
	}
	Assert(spilled == txn->nentries_mem);
	Assert(dlist_is_empty(&txn->changes));
	txn->nentries_mem = 0;
	if (fd != -1)
		CloseTransientFile(fd);
}
static void ReorderBufferSerializeTXN ( ReorderBuffer * rb , ReorderBufferTXN * txn ) { dlist_iter subtxn_i ; dlist_mutable_iter change_i ; int fd = -1 ; XLogSegNo curOpenSegNo = 0 ; Size spilled = 0 ; char path [ MAXPGPATH ] ; elog ( DEBUG2 , " " , ( uint32 ) txn -> nentries_mem , txn -> xid ) ; dlist_foreach ( subtxn_i , & txn -> subtxns ) { ReorderBufferTXN * subtxn ; subtxn = dlist_container ( ReorderBufferTXN , node , subtxn_i . cur ) ; ReorderBufferSerializeTXN ( rb , subtxn ) ; } dlist_foreach_modify ( change_i , & txn -> changes ) { ReorderBufferChange * change ; change = dlist_container ( ReorderBufferChange , node , change_i . cur ) ; if ( fd == -1 || ! XLByteInSeg ( change -> lsn , curOpenSegNo ) ) { XLogRecPtr recptr ; if ( fd != -1 ) CloseTransientFile ( fd ) ; XLByteToSeg ( change -> lsn , curOpenSegNo ) ; XLogSegNoOffsetToRecPtr ( curOpenSegNo , 0 , recptr ) ; sprintf ( path , " " , NameStr ( MyReplicationSlot -> data . name ) , txn -> xid , ( uint32 ) ( recptr >> 32 ) , ( uint32 ) recptr ) ; fd = OpenTransientFile ( path , O_CREAT | O_WRONLY | O_APPEND | PG_BINARY , S_IRUSR | S_IWUSR ) ; if ( fd < 0 ) ereport ( ERROR , ( errcode_for_file_access ( ) , errmsg ( " \" \" " , path ) ) ) ; } ReorderBufferSerializeChange ( rb , txn , fd , change ) ; dlist_delete ( & change -> node ) ; ReorderBufferReturnChange ( rb , change ) ; spilled ++ ; } Assert ( spilled == txn -> nentries_mem ) ; Assert ( dlist_is_empty ( & txn -> changes ) ) ; txn -> nentries_mem = 0 ; if ( fd != -1 ) CloseTransientFile ( fd ) ; }
Spill data of a large transaction (and its subtransactions) to disk.
Spill data of a large transaction ( and its subtransactions ) to disk .
/* do the same to all child TXs */ /* serialize changestream */ /*
		 * store in segment in which it belongs by start lsn, don't split over
		 * multiple segments tho
		 */ /*
			 * No need to care about TLIs here, only used during a single run,
			 * so each LSN only maps to a specific WAL record.
			 */ /* open segment, create it if necessary */