src/backend/replication/logical/reorderbuffer.c
static void
ReorderBufferSerializeChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
							 int fd, ReorderBufferChange *change)
{
	ReorderBufferDiskChange *ondisk;
	Size		sz = sizeof(ReorderBufferDiskChange);

	ReorderBufferSerializeReserve(rb, sz);

	ondisk = (ReorderBufferDiskChange *) rb->outbuf;
	memcpy(&ondisk->change, change, sizeof(ReorderBufferChange));

	switch (change->action)
	{
			/* fall through these, they're all similar enough */
		case REORDER_BUFFER_CHANGE_INSERT:
		case REORDER_BUFFER_CHANGE_UPDATE:
		case REORDER_BUFFER_CHANGE_DELETE:
		case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT:
			{
				char	   *data;
				ReorderBufferTupleBuf *oldtup,
						   *newtup;
				Size		oldlen = 0;
				Size		newlen = 0;

				oldtup = change->data.tp.oldtuple;
				newtup = change->data.tp.newtuple;

				if (oldtup)
				{
					sz += sizeof(HeapTupleData);
					oldlen = oldtup->tuple.t_len;
					sz += oldlen;
				}

				if (newtup)
				{
					sz += sizeof(HeapTupleData);
					newlen = newtup->tuple.t_len;
					sz += newlen;
				}

				/* make sure we have enough space */
				ReorderBufferSerializeReserve(rb, sz);

				data = ((char *) rb->outbuf) + sizeof(ReorderBufferDiskChange);
				/* might have been reallocated above */
				ondisk = (ReorderBufferDiskChange *) rb->outbuf;

				if (oldlen)
				{
					memcpy(data, &oldtup->tuple, sizeof(HeapTupleData));
					data += sizeof(HeapTupleData);

					memcpy(data, oldtup->tuple.t_data, oldlen);
					data += oldlen;
				}

				if (newlen)
				{
					memcpy(data, &newtup->tuple, sizeof(HeapTupleData));
					data += sizeof(HeapTupleData);

					memcpy(data, newtup->tuple.t_data, newlen);
					data += newlen;
				}
				break;
			}
		case REORDER_BUFFER_CHANGE_MESSAGE:
			{
				char	   *data;
				Size		prefix_size = strlen(change->data.msg.prefix) + 1;

				sz += prefix_size + change->data.msg.message_size +
					sizeof(Size) + sizeof(Size);
				ReorderBufferSerializeReserve(rb, sz);

				data = ((char *) rb->outbuf) + sizeof(ReorderBufferDiskChange);

				/* might have been reallocated above */
				ondisk = (ReorderBufferDiskChange *) rb->outbuf;

				/* write the prefix including the size */
				memcpy(data, &prefix_size, sizeof(Size));
				data += sizeof(Size);
				memcpy(data, change->data.msg.prefix,
					   prefix_size);
				data += prefix_size;

				/* write the message including the size */
				memcpy(data, &change->data.msg.message_size, sizeof(Size));
				data += sizeof(Size);
				memcpy(data, change->data.msg.message,
					   change->data.msg.message_size);
				data += change->data.msg.message_size;

				break;
			}
		case REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT:
			{
				Snapshot	snap;
				char	   *data;

				snap = change->data.snapshot;

				sz += sizeof(SnapshotData) +
					sizeof(TransactionId) * snap->xcnt +
					sizeof(TransactionId) * snap->subxcnt
					;

				/* make sure we have enough space */
				ReorderBufferSerializeReserve(rb, sz);
				data = ((char *) rb->outbuf) + sizeof(ReorderBufferDiskChange);
				/* might have been reallocated above */
				ondisk = (ReorderBufferDiskChange *) rb->outbuf;

				memcpy(data, snap, sizeof(SnapshotData));
				data += sizeof(SnapshotData);

				if (snap->xcnt)
				{
					memcpy(data, snap->xip,
						   sizeof(TransactionId) * snap->xcnt);
					data += sizeof(TransactionId) * snap->xcnt;
				}

				if (snap->subxcnt)
				{
					memcpy(data, snap->subxip,
						   sizeof(TransactionId) * snap->subxcnt);
					data += sizeof(TransactionId) * snap->subxcnt;
				}
				break;
			}
		case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM:
		case REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID:
		case REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID:
			/* ReorderBufferChange contains everything important */
			break;
	}

	ondisk->size = sz;

	if (write(fd, rb->outbuf, ondisk->size) != ondisk->size)
	{
		int			save_errno = errno;

		CloseTransientFile(fd);
		errno = save_errno;
		ereport(ERROR,
				(errcode_for_file_access(),
				 errmsg("could not write to data file for XID %u: %m",
						txn->xid)));
	}

	Assert(ondisk->change.action == change->action);
}
/*
 * Serialize individual change to disk.
 */
Serialize individual change to disk.
Serialize individual change to disk .
static void
ReorderBufferSerializeChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
							 int fd, ReorderBufferChange *change)
{
	ReorderBufferDiskChange *ondisk;
	Size		sz = sizeof(ReorderBufferDiskChange);
	ReorderBufferSerializeReserve(rb, sz);
	ondisk = (ReorderBufferDiskChange *) rb->outbuf;
	memcpy(&ondisk->change, change, sizeof(ReorderBufferChange));
	switch (change->action)
	{
		case REORDER_BUFFER_CHANGE_INSERT:
		case REORDER_BUFFER_CHANGE_UPDATE:
		case REORDER_BUFFER_CHANGE_DELETE:
		case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT:
			{
				char	   *data;
				ReorderBufferTupleBuf *oldtup,
						   *newtup;
				Size		oldlen = 0;
				Size		newlen = 0;
				oldtup = change->data.tp.oldtuple;
				newtup = change->data.tp.newtuple;
				if (oldtup)
				{
					sz += sizeof(HeapTupleData);
					oldlen = oldtup->tuple.t_len;
					sz += oldlen;
				}
				if (newtup)
				{
					sz += sizeof(HeapTupleData);
					newlen = newtup->tuple.t_len;
					sz += newlen;
				}
				ReorderBufferSerializeReserve(rb, sz);
				data = ((char *) rb->outbuf) + sizeof(ReorderBufferDiskChange);
				ondisk = (ReorderBufferDiskChange *) rb->outbuf;
				if (oldlen)
				{
					memcpy(data, &oldtup->tuple, sizeof(HeapTupleData));
					data += sizeof(HeapTupleData);
					memcpy(data, oldtup->tuple.t_data, oldlen);
					data += oldlen;
				}
				if (newlen)
				{
					memcpy(data, &newtup->tuple, sizeof(HeapTupleData));
					data += sizeof(HeapTupleData);
					memcpy(data, newtup->tuple.t_data, newlen);
					data += newlen;
				}
				break;
			}
		case REORDER_BUFFER_CHANGE_MESSAGE:
			{
				char	   *data;
				Size		prefix_size = strlen(change->data.msg.prefix) + 1;
				sz += prefix_size + change->data.msg.message_size +
					sizeof(Size) + sizeof(Size);
				ReorderBufferSerializeReserve(rb, sz);
				data = ((char *) rb->outbuf) + sizeof(ReorderBufferDiskChange);
				ondisk = (ReorderBufferDiskChange *) rb->outbuf;
				memcpy(data, &prefix_size, sizeof(Size));
				data += sizeof(Size);
				memcpy(data, change->data.msg.prefix,
					   prefix_size);
				data += prefix_size;
				memcpy(data, &change->data.msg.message_size, sizeof(Size));
				data += sizeof(Size);
				memcpy(data, change->data.msg.message,
					   change->data.msg.message_size);
				data += change->data.msg.message_size;
				break;
			}
		case REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT:
			{
				Snapshot	snap;
				char	   *data;
				snap = change->data.snapshot;
				sz += sizeof(SnapshotData) +
					sizeof(TransactionId) * snap->xcnt +
					sizeof(TransactionId) * snap->subxcnt
					;
				ReorderBufferSerializeReserve(rb, sz);
				data = ((char *) rb->outbuf) + sizeof(ReorderBufferDiskChange);
				ondisk = (ReorderBufferDiskChange *) rb->outbuf;
				memcpy(data, snap, sizeof(SnapshotData));
				data += sizeof(SnapshotData);
				if (snap->xcnt)
				{
					memcpy(data, snap->xip,
						   sizeof(TransactionId) * snap->xcnt);
					data += sizeof(TransactionId) * snap->xcnt;
				}
				if (snap->subxcnt)
				{
					memcpy(data, snap->subxip,
						   sizeof(TransactionId) * snap->subxcnt);
					data += sizeof(TransactionId) * snap->subxcnt;
				}
				break;
			}
		case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM:
		case REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID:
		case REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID:
			break;
	}
	ondisk->size = sz;
	if (write(fd, rb->outbuf, ondisk->size) != ondisk->size)
	{
		int			save_errno = errno;
		CloseTransientFile(fd);
		errno = save_errno;
		ereport(ERROR,
				(errcode_for_file_access(),
				 errmsg("could not write to data file for XID %u: %m",
						txn->xid)));
	}
	Assert(ondisk->change.action == change->action);
}
static void ReorderBufferSerializeChange ( ReorderBuffer * rb , ReorderBufferTXN * txn , int fd , ReorderBufferChange * change ) { ReorderBufferDiskChange * ondisk ; Size sz = sizeof ( ReorderBufferDiskChange ) ; ReorderBufferSerializeReserve ( rb , sz ) ; ondisk = ( ReorderBufferDiskChange * ) rb -> outbuf ; memcpy ( & ondisk -> change , change , sizeof ( ReorderBufferChange ) ) ; switch ( change -> action ) { case REORDER_BUFFER_CHANGE_INSERT : case REORDER_BUFFER_CHANGE_UPDATE : case REORDER_BUFFER_CHANGE_DELETE : case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT : { char * data ; ReorderBufferTupleBuf * oldtup , * newtup ; Size oldlen = 0 ; Size newlen = 0 ; oldtup = change -> data . tp . oldtuple ; newtup = change -> data . tp . newtuple ; if ( oldtup ) { sz += sizeof ( HeapTupleData ) ; oldlen = oldtup -> tuple . t_len ; sz += oldlen ; } if ( newtup ) { sz += sizeof ( HeapTupleData ) ; newlen = newtup -> tuple . t_len ; sz += newlen ; } ReorderBufferSerializeReserve ( rb , sz ) ; data = ( ( char * ) rb -> outbuf ) + sizeof ( ReorderBufferDiskChange ) ; ondisk = ( ReorderBufferDiskChange * ) rb -> outbuf ; if ( oldlen ) { memcpy ( data , & oldtup -> tuple , sizeof ( HeapTupleData ) ) ; data += sizeof ( HeapTupleData ) ; memcpy ( data , oldtup -> tuple . t_data , oldlen ) ; data += oldlen ; } if ( newlen ) { memcpy ( data , & newtup -> tuple , sizeof ( HeapTupleData ) ) ; data += sizeof ( HeapTupleData ) ; memcpy ( data , newtup -> tuple . t_data , newlen ) ; data += newlen ; } break ; } case REORDER_BUFFER_CHANGE_MESSAGE : { char * data ; Size prefix_size = strlen ( change -> data . msg . prefix ) + 1 ; sz += prefix_size + change -> data . msg . message_size + sizeof ( Size ) + sizeof ( Size ) ; ReorderBufferSerializeReserve ( rb , sz ) ; data = ( ( char * ) rb -> outbuf ) + sizeof ( ReorderBufferDiskChange ) ; ondisk = ( ReorderBufferDiskChange * ) rb -> outbuf ; memcpy ( data , & prefix_size , sizeof ( Size ) ) ; data += sizeof ( Size ) ; memcpy ( data , change -> data . msg . prefix , prefix_size ) ; data += prefix_size ; memcpy ( data , & change -> data . msg . message_size , sizeof ( Size ) ) ; data += sizeof ( Size ) ; memcpy ( data , change -> data . msg . message , change -> data . msg . message_size ) ; data += change -> data . msg . message_size ; break ; } case REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT : { Snapshot snap ; char * data ; snap = change -> data . snapshot ; sz += sizeof ( SnapshotData ) + sizeof ( TransactionId ) * snap -> xcnt + sizeof ( TransactionId ) * snap -> subxcnt ; ReorderBufferSerializeReserve ( rb , sz ) ; data = ( ( char * ) rb -> outbuf ) + sizeof ( ReorderBufferDiskChange ) ; ondisk = ( ReorderBufferDiskChange * ) rb -> outbuf ; memcpy ( data , snap , sizeof ( SnapshotData ) ) ; data += sizeof ( SnapshotData ) ; if ( snap -> xcnt ) { memcpy ( data , snap -> xip , sizeof ( TransactionId ) * snap -> xcnt ) ; data += sizeof ( TransactionId ) * snap -> xcnt ; } if ( snap -> subxcnt ) { memcpy ( data , snap -> subxip , sizeof ( TransactionId ) * snap -> subxcnt ) ; data += sizeof ( TransactionId ) * snap -> subxcnt ; } break ; } case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM : case REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID : case REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID : break ; } ondisk -> size = sz ; if ( write ( fd , rb -> outbuf , ondisk -> size ) != ondisk -> size ) { int save_errno = errno ; CloseTransientFile ( fd ) ; errno = save_errno ; ereport ( ERROR , ( errcode_for_file_access ( ) , errmsg ( " " , txn -> xid ) ) ) ; } Assert ( ondisk -> change . action == change -> action ) ; }
Serialize individual change to disk.
Serialize individual change to disk .
/* fall through these, they're all similar enough */ /* make sure we have enough space */ /* might have been reallocated above */ /* might have been reallocated above */ /* write the prefix including the size */ /* write the message including the size */ /* make sure we have enough space */ /* might have been reallocated above */ /* ReorderBufferChange contains everything important */