src/backend/replication/logical/reorderbuffer.c
static Size
ReorderBufferRestoreChanges(ReorderBuffer *rb, ReorderBufferTXN *txn,
							int *fd, XLogSegNo *segno)
{
	Size		restored = 0;
	XLogSegNo	last_segno;
	dlist_mutable_iter cleanup_iter;

	Assert(txn->first_lsn != InvalidXLogRecPtr);
	Assert(txn->final_lsn != InvalidXLogRecPtr);

	/* free current entries, so we have memory for more */
	dlist_foreach_modify(cleanup_iter, &txn->changes)
	{
		ReorderBufferChange *cleanup =
		dlist_container(ReorderBufferChange, node, cleanup_iter.cur);

		dlist_delete(&cleanup->node);
		ReorderBufferReturnChange(rb, cleanup);
	}
	txn->nentries_mem = 0;
	Assert(dlist_is_empty(&txn->changes));

	XLByteToSeg(txn->final_lsn, last_segno);

	while (restored < max_changes_in_memory && *segno <= last_segno)
	{
		int			readBytes;
		ReorderBufferDiskChange *ondisk;

		if (*fd == -1)
		{
			XLogRecPtr	recptr;
			char		path[MAXPGPATH];

			/* first time in */
			if (*segno == 0)
			{
				XLByteToSeg(txn->first_lsn, *segno);
			}

			Assert(*segno != 0 || dlist_is_empty(&txn->changes));
			XLogSegNoOffsetToRecPtr(*segno, 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);

			*fd = OpenTransientFile(path, O_RDONLY | PG_BINARY, 0);
			if (*fd < 0 && errno == ENOENT)
			{
				*fd = -1;
				(*segno)++;
				continue;
			}
			else if (*fd < 0)
				ereport(ERROR,
						(errcode_for_file_access(),
						 errmsg("could not open file \"%s\": %m",
								path)));

		}

		/*
		 * Read the statically sized part of a change which has information
		 * about the total size. If we couldn't read a record, we're at the
		 * end of this file.
		 */
		ReorderBufferSerializeReserve(rb, sizeof(ReorderBufferDiskChange));
		readBytes = read(*fd, rb->outbuf, sizeof(ReorderBufferDiskChange));

		/* eof */
		if (readBytes == 0)
		{
			CloseTransientFile(*fd);
			*fd = -1;
			(*segno)++;
			continue;
		}
		else if (readBytes < 0)
			ereport(ERROR,
					(errcode_for_file_access(),
				errmsg("could not read from reorderbuffer spill file: %m")));
		else if (readBytes != sizeof(ReorderBufferDiskChange))
			ereport(ERROR,
					(errcode_for_file_access(),
					 errmsg("could not read from reorderbuffer spill file: read %d instead of %u bytes",
							readBytes,
							(uint32) sizeof(ReorderBufferDiskChange))));

		ondisk = (ReorderBufferDiskChange *) rb->outbuf;

		ReorderBufferSerializeReserve(rb,
							 sizeof(ReorderBufferDiskChange) + ondisk->size);
		ondisk = (ReorderBufferDiskChange *) rb->outbuf;

		readBytes = read(*fd, rb->outbuf + sizeof(ReorderBufferDiskChange),
						 ondisk->size - sizeof(ReorderBufferDiskChange));

		if (readBytes < 0)
			ereport(ERROR,
					(errcode_for_file_access(),
				errmsg("could not read from reorderbuffer spill file: %m")));
		else if (readBytes != ondisk->size - sizeof(ReorderBufferDiskChange))
			ereport(ERROR,
					(errcode_for_file_access(),
					 errmsg("could not read from reorderbuffer spill file: read %d instead of %u bytes",
							readBytes,
				(uint32) (ondisk->size - sizeof(ReorderBufferDiskChange)))));

		/*
		 * ok, read a full change from disk, now restore it into proper
		 * in-memory format
		 */
		ReorderBufferRestoreChange(rb, txn, rb->outbuf);
		restored++;
	}

	return restored;
}
/*
 * Restore a number of changes spilled to disk back into memory.
 */
Restore a number of changes spilled to disk back into memory.
Restore a number of changes spilled to disk back into memory .
static Size
ReorderBufferRestoreChanges(ReorderBuffer *rb, ReorderBufferTXN *txn,
							int *fd, XLogSegNo *segno)
{
	Size		restored = 0;
	XLogSegNo	last_segno;
	dlist_mutable_iter cleanup_iter;
	Assert(txn->first_lsn != InvalidXLogRecPtr);
	Assert(txn->final_lsn != InvalidXLogRecPtr);
	dlist_foreach_modify(cleanup_iter, &txn->changes)
	{
		ReorderBufferChange *cleanup =
		dlist_container(ReorderBufferChange, node, cleanup_iter.cur);
		dlist_delete(&cleanup->node);
		ReorderBufferReturnChange(rb, cleanup);
	}
	txn->nentries_mem = 0;
	Assert(dlist_is_empty(&txn->changes));
	XLByteToSeg(txn->final_lsn, last_segno);
	while (restored < max_changes_in_memory && *segno <= last_segno)
	{
		int			readBytes;
		ReorderBufferDiskChange *ondisk;
		if (*fd == -1)
		{
			XLogRecPtr	recptr;
			char		path[MAXPGPATH];
			if (*segno == 0)
			{
				XLByteToSeg(txn->first_lsn, *segno);
			}
			Assert(*segno != 0 || dlist_is_empty(&txn->changes));
			XLogSegNoOffsetToRecPtr(*segno, 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_RDONLY | PG_BINARY, 0);
			if (*fd < 0 && errno == ENOENT)
			{
				*fd = -1;
				(*segno)++;
				continue;
			}
			else if (*fd < 0)
				ereport(ERROR,
						(errcode_for_file_access(),
						 errmsg("could not open file \"%s\": %m",
								path)));
		}
		ReorderBufferSerializeReserve(rb, sizeof(ReorderBufferDiskChange));
		readBytes = read(*fd, rb->outbuf, sizeof(ReorderBufferDiskChange));
		if (readBytes == 0)
		{
			CloseTransientFile(*fd);
			*fd = -1;
			(*segno)++;
			continue;
		}
		else if (readBytes < 0)
			ereport(ERROR,
					(errcode_for_file_access(),
				errmsg("could not read from reorderbuffer spill file: %m")));
		else if (readBytes != sizeof(ReorderBufferDiskChange))
			ereport(ERROR,
					(errcode_for_file_access(),
					 errmsg("could not read from reorderbuffer spill file: read %d instead of %u bytes",
							readBytes,
							(uint32) sizeof(ReorderBufferDiskChange))));
		ondisk = (ReorderBufferDiskChange *) rb->outbuf;
		ReorderBufferSerializeReserve(rb,
							 sizeof(ReorderBufferDiskChange) + ondisk->size);
		ondisk = (ReorderBufferDiskChange *) rb->outbuf;
		readBytes = read(*fd, rb->outbuf + sizeof(ReorderBufferDiskChange),
						 ondisk->size - sizeof(ReorderBufferDiskChange));
		if (readBytes < 0)
			ereport(ERROR,
					(errcode_for_file_access(),
				errmsg("could not read from reorderbuffer spill file: %m")));
		else if (readBytes != ondisk->size - sizeof(ReorderBufferDiskChange))
			ereport(ERROR,
					(errcode_for_file_access(),
					 errmsg("could not read from reorderbuffer spill file: read %d instead of %u bytes",
							readBytes,
				(uint32) (ondisk->size - sizeof(ReorderBufferDiskChange)))));
		ReorderBufferRestoreChange(rb, txn, rb->outbuf);
		restored++;
	}
	return restored;
}
static Size ReorderBufferRestoreChanges ( ReorderBuffer * rb , ReorderBufferTXN * txn , int * fd , XLogSegNo * segno ) { Size restored = 0 ; XLogSegNo last_segno ; dlist_mutable_iter cleanup_iter ; Assert ( txn -> first_lsn != InvalidXLogRecPtr ) ; Assert ( txn -> final_lsn != InvalidXLogRecPtr ) ; dlist_foreach_modify ( cleanup_iter , & txn -> changes ) { ReorderBufferChange * cleanup = dlist_container ( ReorderBufferChange , node , cleanup_iter . cur ) ; dlist_delete ( & cleanup -> node ) ; ReorderBufferReturnChange ( rb , cleanup ) ; } txn -> nentries_mem = 0 ; Assert ( dlist_is_empty ( & txn -> changes ) ) ; XLByteToSeg ( txn -> final_lsn , last_segno ) ; while ( restored < max_changes_in_memory && * segno <= last_segno ) { int readBytes ; ReorderBufferDiskChange * ondisk ; if ( * fd == -1 ) { XLogRecPtr recptr ; char path [ MAXPGPATH ] ; if ( * segno == 0 ) { XLByteToSeg ( txn -> first_lsn , * segno ) ; } Assert ( * segno != 0 || dlist_is_empty ( & txn -> changes ) ) ; XLogSegNoOffsetToRecPtr ( * segno , 0 , recptr ) ; sprintf ( path , " " , NameStr ( MyReplicationSlot -> data . name ) , txn -> xid , ( uint32 ) ( recptr >> 32 ) , ( uint32 ) recptr ) ; * fd = OpenTransientFile ( path , O_RDONLY | PG_BINARY , 0 ) ; if ( * fd < 0 && errno == ENOENT ) { * fd = -1 ; ( * segno ) ++ ; continue ; } else if ( * fd < 0 ) ereport ( ERROR , ( errcode_for_file_access ( ) , errmsg ( " \" \" " , path ) ) ) ; } ReorderBufferSerializeReserve ( rb , sizeof ( ReorderBufferDiskChange ) ) ; readBytes = read ( * fd , rb -> outbuf , sizeof ( ReorderBufferDiskChange ) ) ; if ( readBytes == 0 ) { CloseTransientFile ( * fd ) ; * fd = -1 ; ( * segno ) ++ ; continue ; } else if ( readBytes < 0 ) ereport ( ERROR , ( errcode_for_file_access ( ) , errmsg ( " " ) ) ) ; else if ( readBytes != sizeof ( ReorderBufferDiskChange ) ) ereport ( ERROR , ( errcode_for_file_access ( ) , errmsg ( " " , readBytes , ( uint32 ) sizeof ( ReorderBufferDiskChange ) ) ) ) ; ondisk = ( ReorderBufferDiskChange * ) rb -> outbuf ; ReorderBufferSerializeReserve ( rb , sizeof ( ReorderBufferDiskChange ) + ondisk -> size ) ; ondisk = ( ReorderBufferDiskChange * ) rb -> outbuf ; readBytes = read ( * fd , rb -> outbuf + sizeof ( ReorderBufferDiskChange ) , ondisk -> size - sizeof ( ReorderBufferDiskChange ) ) ; if ( readBytes < 0 ) ereport ( ERROR , ( errcode_for_file_access ( ) , errmsg ( " " ) ) ) ; else if ( readBytes != ondisk -> size - sizeof ( ReorderBufferDiskChange ) ) ereport ( ERROR , ( errcode_for_file_access ( ) , errmsg ( " " , readBytes , ( uint32 ) ( ondisk -> size - sizeof ( ReorderBufferDiskChange ) ) ) ) ) ; ReorderBufferRestoreChange ( rb , txn , rb -> outbuf ) ; restored ++ ; } return restored ; }
Restore a number of changes spilled to disk back into memory.
Restore a number of changes spilled to disk back into memory .
/* free current entries, so we have memory for more */ /* first time in */ /*
			 * No need to care about TLIs here, only used during a single run,
			 * so each LSN only maps to a specific WAL record.
			 */ /*
		 * Read the statically sized part of a change which has information
		 * about the total size. If we couldn't read a record, we're at the
		 * end of this file.
		 */ /* eof */ /*
		 * ok, read a full change from disk, now restore it into proper
		 * in-memory format
		 */