Function apply_handle_stream_commit

src/backend/replication/logical/worker.c:1003–1144

* Handle STREAM COMMIT message. */

 * Handle STREAM COMMIT message.
 */
static void
apply_handle_stream_commit(StringInfo s)
{
	TransactionId xid;
	StringInfoData s2;
	int			nchanges;
	char		path[MAXPGPATH];
	char	   *buffer = NULL;
	LogicalRepCommitData commit_data;
	StreamXidHash *ent;
	MemoryContext oldcxt;
	BufFile    *fd;

	if (in_streamed_transaction)
		ereport(ERROR,
				(errcode(ERRCODE_PROTOCOL_VIOLATION),
				 errmsg_internal("STREAM COMMIT message without STREAM STOP")));

	xid = logicalrep_read_stream_commit(s, &commit_data);

	elog(DEBUG1, "received commit for streamed transaction %u", xid);

	/* Make sure we have an open transaction */
	begin_replication_step();

	/*
	 * Allocate file handle and memory required to process all the messages in
	 * TopTransactionContext to avoid them getting reset after each message is
	 * processed.
	 */
	oldcxt = MemoryContextSwitchTo(TopTransactionContext);

	/* open the spool file for the committed transaction */
	changes_filename(path, MyLogicalRepWorker->subid, xid);
	elog(DEBUG1, "replaying changes from file \"%s\"", path);

	ent = (StreamXidHash *) hash_search(xidhash,
										(void *) &xid,
										HASH_FIND,
										NULL);
	if (!ent)
		ereport(ERROR,
				(errcode(ERRCODE_PROTOCOL_VIOLATION),
				 errmsg_internal("transaction %u not found in stream XID hash table",
								 xid)));

	fd = BufFileOpenShared(ent->stream_fileset, path, O_RDONLY);

	buffer = palloc(BLCKSZ);
	initStringInfo(&s2);

	MemoryContextSwitchTo(oldcxt);

	remote_final_lsn = commit_data.commit_lsn;

	/*
	 * Make sure the handle apply_dispatch methods are aware we're in a remote
	 * transaction.