src/backend/replication/logical/reorderbuffer.c
void
ReorderBufferCommit(ReorderBuffer *rb, TransactionId xid,
					XLogRecPtr commit_lsn, XLogRecPtr end_lsn,
					TimestampTz commit_time,
					RepOriginId origin_id, XLogRecPtr origin_lsn)
{
	ReorderBufferTXN *txn;
	volatile Snapshot snapshot_now;
	volatile CommandId command_id = FirstCommandId;
	bool		using_subtxn;
	ReorderBufferIterTXNState *volatile iterstate = NULL;

	txn = ReorderBufferTXNByXid(rb, xid, false, NULL, InvalidXLogRecPtr,
								false);

	/* unknown transaction, nothing to replay */
	if (txn == NULL)
		return;

	txn->final_lsn = commit_lsn;
	txn->end_lsn = end_lsn;
	txn->commit_time = commit_time;
	txn->origin_id = origin_id;
	txn->origin_lsn = origin_lsn;

	/*
	 * If this transaction didn't have any real changes in our database, it's
	 * OK not to have a snapshot. Note that ReorderBufferCommitChild will have
	 * transferred its snapshot to this transaction if it had one and the
	 * toplevel tx didn't.
	 */
	if (txn->base_snapshot == NULL)
	{
		Assert(txn->ninvalidations == 0);
		ReorderBufferCleanupTXN(rb, txn);
		return;
	}

	snapshot_now = txn->base_snapshot;

	/* build data to be able to lookup the CommandIds of catalog tuples */
	ReorderBufferBuildTupleCidHash(rb, txn);

	/* setup the initial snapshot */
	SetupHistoricSnapshot(snapshot_now, txn->tuplecid_hash);

	/*
	 * Decoding needs access to syscaches et al., which in turn use
	 * heavyweight locks and such. Thus we need to have enough state around to
	 * keep track of those.  The easiest way is to simply use a transaction
	 * internally.  That also allows us to easily enforce that nothing writes
	 * to the database by checking for xid assignments.
	 *
	 * When we're called via the SQL SRF there's already a transaction
	 * started, so start an explicit subtransaction there.
	 */
	using_subtxn = IsTransactionOrTransactionBlock();

	PG_TRY();
	{
		ReorderBufferChange *change;
		ReorderBufferChange *specinsert = NULL;

		if (using_subtxn)
			BeginInternalSubTransaction("replay");
		else
			StartTransactionCommand();

		rb->begin(rb, txn);

		iterstate = ReorderBufferIterTXNInit(rb, txn);
		while ((change = ReorderBufferIterTXNNext(rb, iterstate)) != NULL)
		{
			Relation	relation = NULL;
			Oid			reloid;

			switch (change->action)
			{
				case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM:

					/*
					 * Confirmation for speculative insertion arrived. Simply
					 * use as a normal record. It'll be cleaned up at the end
					 * of INSERT processing.
					 */
					Assert(specinsert->data.tp.oldtuple == NULL);
					change = specinsert;
					change->action = REORDER_BUFFER_CHANGE_INSERT;

					/* intentionally fall through */
				case REORDER_BUFFER_CHANGE_INSERT:
				case REORDER_BUFFER_CHANGE_UPDATE:
				case REORDER_BUFFER_CHANGE_DELETE:
					Assert(snapshot_now);

					reloid = RelidByRelfilenode(change->data.tp.relnode.spcNode,
											change->data.tp.relnode.relNode);

					/*
					 * Catalog tuple without data, emitted while catalog was
					 * in the process of being rewritten.
					 */
					if (reloid == InvalidOid &&
						change->data.tp.newtuple == NULL &&
						change->data.tp.oldtuple == NULL)
						goto change_done;
					else if (reloid == InvalidOid)
						elog(ERROR, "could not map filenode \"%s\" to relation OID",
							 relpathperm(change->data.tp.relnode,
										 MAIN_FORKNUM));

					relation = RelationIdGetRelation(reloid);

					if (relation == NULL)
						elog(ERROR, "could not open relation with OID %u (for filenode \"%s\")",
							 reloid,
							 relpathperm(change->data.tp.relnode,
										 MAIN_FORKNUM));

					if (!RelationIsLogicallyLogged(relation))
						goto change_done;

					/*
					 * For now ignore sequence changes entirely. Most of the
					 * time they don't log changes using records we
					 * understand, so it doesn't make sense to handle the few
					 * cases we do.
					 */
					if (relation->rd_rel->relkind == RELKIND_SEQUENCE)
						goto change_done;

					/* user-triggered change */
					if (!IsToastRelation(relation))
					{
						ReorderBufferToastReplace(rb, txn, relation, change);
						rb->apply_change(rb, txn, relation, change);

						/*
						 * Only clear reassembled toast chunks if we're sure
						 * they're not required anymore. The creator of the
						 * tuple tells us.
						 */
						if (change->data.tp.clear_toast_afterwards)
							ReorderBufferToastReset(rb, txn);
					}
					/* we're not interested in toast deletions */
					else if (change->action == REORDER_BUFFER_CHANGE_INSERT)
					{
						/*
						 * Need to reassemble the full toasted Datum in
						 * memory, to ensure the chunks don't get reused till
						 * we're done remove it from the list of this
						 * transaction's changes. Otherwise it will get
						 * freed/reused while restoring spooled data from
						 * disk.
						 */
						dlist_delete(&change->node);
						ReorderBufferToastAppendChunk(rb, txn, relation,
													  change);
					}

			change_done:

					/*
					 * Either speculative insertion was confirmed, or it was
					 * unsuccessful and the record isn't needed anymore.
					 */
					if (specinsert != NULL)
					{
						ReorderBufferReturnChange(rb, specinsert);
						specinsert = NULL;
					}

					if (relation != NULL)
					{
						RelationClose(relation);
						relation = NULL;
					}
					break;

				case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT:

					/*
					 * Speculative insertions are dealt with by delaying the
					 * processing of the insert until the confirmation record
					 * arrives. For that we simply unlink the record from the
					 * chain, so it does not get freed/reused while restoring
					 * spooled data from disk.
					 *
					 * This is safe in the face of concurrent catalog changes
					 * because the relevant relation can't be changed between
					 * speculative insertion and confirmation due to
					 * CheckTableNotInUse() and locking.
					 */

					/* clear out a pending (and thus failed) speculation */
					if (specinsert != NULL)
					{
						ReorderBufferReturnChange(rb, specinsert);
						specinsert = NULL;
					}

					/* and memorize the pending insertion */
					dlist_delete(&change->node);
					specinsert = change;
					break;

				case REORDER_BUFFER_CHANGE_MESSAGE:
					rb->message(rb, txn, change->lsn, true,
								change->data.msg.prefix,
								change->data.msg.message_size,
								change->data.msg.message);
					break;

				case REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT:
					/* get rid of the old */
					TeardownHistoricSnapshot(false);

					if (snapshot_now->copied)
					{
						ReorderBufferFreeSnap(rb, snapshot_now);
						snapshot_now =
							ReorderBufferCopySnap(rb, change->data.snapshot,
												  txn, command_id);
					}

					/*
					 * Restored from disk, need to be careful not to double
					 * free. We could introduce refcounting for that, but for
					 * now this seems infrequent enough not to care.
					 */
					else if (change->data.snapshot->copied)
					{
						snapshot_now =
							ReorderBufferCopySnap(rb, change->data.snapshot,
												  txn, command_id);
					}
					else
					{
						snapshot_now = change->data.snapshot;
					}


					/* and continue with the new one */
					SetupHistoricSnapshot(snapshot_now, txn->tuplecid_hash);
					break;

				case REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID:
					Assert(change->data.command_id != InvalidCommandId);

					if (command_id < change->data.command_id)
					{
						command_id = change->data.command_id;

						if (!snapshot_now->copied)
						{
							/* we don't use the global one anymore */
							snapshot_now = ReorderBufferCopySnap(rb, snapshot_now,
															txn, command_id);
						}

						snapshot_now->curcid = command_id;

						TeardownHistoricSnapshot(false);
						SetupHistoricSnapshot(snapshot_now, txn->tuplecid_hash);

						/*
						 * Every time the CommandId is incremented, we could
						 * see new catalog contents, so execute all
						 * invalidations.
						 */
						ReorderBufferExecuteInvalidations(rb, txn);
					}

					break;

				case REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID:
					elog(ERROR, "tuplecid value in changequeue");
					break;
			}
		}

		/*
		 * There's a speculative insertion remaining, just clean in up, it
		 * can't have been successful, otherwise we'd gotten a confirmation
		 * record.
		 */
		if (specinsert)
		{
			ReorderBufferReturnChange(rb, specinsert);
			specinsert = NULL;
		}

		/* clean up the iterator */
		ReorderBufferIterTXNFinish(rb, iterstate);
		iterstate = NULL;

		/* call commit callback */
		rb->commit(rb, txn, commit_lsn);

		/* this is just a sanity check against bad output plugin behaviour */
		if (GetCurrentTransactionIdIfAny() != InvalidTransactionId)
			elog(ERROR, "output plugin used XID %u",
				 GetCurrentTransactionId());

		/* cleanup */
		TeardownHistoricSnapshot(false);

		/*
		 * Aborting the current (sub-)transaction as a whole has the right
		 * semantics. We want all locks acquired in here to be released, not
		 * reassigned to the parent and we do not want any database access
		 * have persistent effects.
		 */
		AbortCurrentTransaction();

		/* make sure there's no cache pollution */
		ReorderBufferExecuteInvalidations(rb, txn);

		if (using_subtxn)
			RollbackAndReleaseCurrentSubTransaction();

		if (snapshot_now->copied)
			ReorderBufferFreeSnap(rb, snapshot_now);

		/* remove potential on-disk data, and deallocate */
		ReorderBufferCleanupTXN(rb, txn);
	}
	PG_CATCH();
	{
		/* TODO: Encapsulate cleanup from the PG_TRY and PG_CATCH blocks */
		if (iterstate)
			ReorderBufferIterTXNFinish(rb, iterstate);

		TeardownHistoricSnapshot(true);

		/*
		 * Force cache invalidation to happen outside of a valid transaction
		 * to prevent catalog access as we just caught an error.
		 */
		AbortCurrentTransaction();

		/* make sure there's no cache pollution */
		ReorderBufferExecuteInvalidations(rb, txn);

		if (using_subtxn)
			RollbackAndReleaseCurrentSubTransaction();

		if (snapshot_now->copied)
			ReorderBufferFreeSnap(rb, snapshot_now);

		/* remove potential on-disk data, and deallocate */
		ReorderBufferCleanupTXN(rb, txn);

		PG_RE_THROW();
	}
	PG_END_TRY();
}
/*
 * Perform the replay of a transaction and it's non-aborted subtransactions.
 *
 * Subtransactions previously have to be processed by
 * ReorderBufferCommitChild(), even if previously assigned to the toplevel
 * transaction with ReorderBufferAssignChild.
 *
 * We currently can only decode a transaction's contents in when their commit
 * record is read because that's currently the only place where we know about
 * cache invalidations. Thus, once a toplevel commit is read, we iterate over
 * the top and subtransactions (using a k-way merge) and replay the changes in
 * lsn order.
 */
Perform the replay of a transaction and it's non-aborted subtransactions.
Subtransactions previously have to be processed by
ReorderBufferCommitChild(), even if previously assigned to the toplevel
transaction with ReorderBufferAssignChild.

We currently can only decode a transaction's contents in when their commit
record is read because that's currently the only place where we know about
cache invalidations. Thus, once a toplevel commit is read, we iterate over
the top and subtransactions (using a k-way merge) and replay the changes in
lsn order.
Perform the replay of a transaction and it ' s non - aborted subtransactions . Subtransactions previously have to be processed by ReorderBufferCommitChild () even if previously assigned to the toplevel transaction with ReorderBufferAssignChild . We currently can only decode a transaction ' s contents in when their commit record is read because that ' s currently the only place where we know about cache invalidations . Thus once a toplevel commit is read we iterate over the top and subtransactions ( using a k - way merge ) and replay the changes in lsn order .
void
ReorderBufferCommit(ReorderBuffer *rb, TransactionId xid,
					XLogRecPtr commit_lsn, XLogRecPtr end_lsn,
					TimestampTz commit_time,
					RepOriginId origin_id, XLogRecPtr origin_lsn)
{
	ReorderBufferTXN *txn;
	volatile Snapshot snapshot_now;
	volatile CommandId command_id = FirstCommandId;
	bool		using_subtxn;
	ReorderBufferIterTXNState *volatile iterstate = NULL;
	txn = ReorderBufferTXNByXid(rb, xid, false, NULL, InvalidXLogRecPtr,
								false);
	if (txn == NULL)
		return;
	txn->final_lsn = commit_lsn;
	txn->end_lsn = end_lsn;
	txn->commit_time = commit_time;
	txn->origin_id = origin_id;
	txn->origin_lsn = origin_lsn;
	if (txn->base_snapshot == NULL)
	{
		Assert(txn->ninvalidations == 0);
		ReorderBufferCleanupTXN(rb, txn);
		return;
	}
	snapshot_now = txn->base_snapshot;
	ReorderBufferBuildTupleCidHash(rb, txn);
	SetupHistoricSnapshot(snapshot_now, txn->tuplecid_hash);
	using_subtxn = IsTransactionOrTransactionBlock();
	PG_TRY();
	{
		ReorderBufferChange *change;
		ReorderBufferChange *specinsert = NULL;
		if (using_subtxn)
			BeginInternalSubTransaction("replay");
		else
			StartTransactionCommand();
		rb->begin(rb, txn);
		iterstate = ReorderBufferIterTXNInit(rb, txn);
		while ((change = ReorderBufferIterTXNNext(rb, iterstate)) != NULL)
		{
			Relation	relation = NULL;
			Oid			reloid;
			switch (change->action)
			{
				case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM:
					Assert(specinsert->data.tp.oldtuple == NULL);
					change = specinsert;
					change->action = REORDER_BUFFER_CHANGE_INSERT;
				case REORDER_BUFFER_CHANGE_INSERT:
				case REORDER_BUFFER_CHANGE_UPDATE:
				case REORDER_BUFFER_CHANGE_DELETE:
					Assert(snapshot_now);
					reloid = RelidByRelfilenode(change->data.tp.relnode.spcNode,
											change->data.tp.relnode.relNode);
					if (reloid == InvalidOid &&
						change->data.tp.newtuple == NULL &&
						change->data.tp.oldtuple == NULL)
						goto change_done;
					else if (reloid == InvalidOid)
						elog(ERROR, "could not map filenode \"%s\" to relation OID",
							 relpathperm(change->data.tp.relnode,
										 MAIN_FORKNUM));
					relation = RelationIdGetRelation(reloid);
					if (relation == NULL)
						elog(ERROR, "could not open relation with OID %u (for filenode \"%s\")",
							 reloid,
							 relpathperm(change->data.tp.relnode,
										 MAIN_FORKNUM));
					if (!RelationIsLogicallyLogged(relation))
						goto change_done;
					if (relation->rd_rel->relkind == RELKIND_SEQUENCE)
						goto change_done;
					if (!IsToastRelation(relation))
					{
						ReorderBufferToastReplace(rb, txn, relation, change);
						rb->apply_change(rb, txn, relation, change);
						if (change->data.tp.clear_toast_afterwards)
							ReorderBufferToastReset(rb, txn);
					}
					else if (change->action == REORDER_BUFFER_CHANGE_INSERT)
					{
						dlist_delete(&change->node);
						ReorderBufferToastAppendChunk(rb, txn, relation,
													  change);
					}
			change_done:
					if (specinsert != NULL)
					{
						ReorderBufferReturnChange(rb, specinsert);
						specinsert = NULL;
					}
					if (relation != NULL)
					{
						RelationClose(relation);
						relation = NULL;
					}
					break;
				case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT:
					if (specinsert != NULL)
					{
						ReorderBufferReturnChange(rb, specinsert);
						specinsert = NULL;
					}
					dlist_delete(&change->node);
					specinsert = change;
					break;
				case REORDER_BUFFER_CHANGE_MESSAGE:
					rb->message(rb, txn, change->lsn, true,
								change->data.msg.prefix,
								change->data.msg.message_size,
								change->data.msg.message);
					break;
				case REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT:
					TeardownHistoricSnapshot(false);
					if (snapshot_now->copied)
					{
						ReorderBufferFreeSnap(rb, snapshot_now);
						snapshot_now =
							ReorderBufferCopySnap(rb, change->data.snapshot,
												  txn, command_id);
					}
					else if (change->data.snapshot->copied)
					{
						snapshot_now =
							ReorderBufferCopySnap(rb, change->data.snapshot,
												  txn, command_id);
					}
					else
					{
						snapshot_now = change->data.snapshot;
					}
					SetupHistoricSnapshot(snapshot_now, txn->tuplecid_hash);
					break;
				case REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID:
					Assert(change->data.command_id != InvalidCommandId);
					if (command_id < change->data.command_id)
					{
						command_id = change->data.command_id;
						if (!snapshot_now->copied)
						{
							snapshot_now = ReorderBufferCopySnap(rb, snapshot_now,
															txn, command_id);
						}
						snapshot_now->curcid = command_id;
						TeardownHistoricSnapshot(false);
						SetupHistoricSnapshot(snapshot_now, txn->tuplecid_hash);
						ReorderBufferExecuteInvalidations(rb, txn);
					}
					break;
				case REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID:
					elog(ERROR, "tuplecid value in changequeue");
					break;
			}
		}
		if (specinsert)
		{
			ReorderBufferReturnChange(rb, specinsert);
			specinsert = NULL;
		}
		ReorderBufferIterTXNFinish(rb, iterstate);
		iterstate = NULL;
		rb->commit(rb, txn, commit_lsn);
		if (GetCurrentTransactionIdIfAny() != InvalidTransactionId)
			elog(ERROR, "output plugin used XID %u",
				 GetCurrentTransactionId());
		TeardownHistoricSnapshot(false);
		AbortCurrentTransaction();
		ReorderBufferExecuteInvalidations(rb, txn);
		if (using_subtxn)
			RollbackAndReleaseCurrentSubTransaction();
		if (snapshot_now->copied)
			ReorderBufferFreeSnap(rb, snapshot_now);
		ReorderBufferCleanupTXN(rb, txn);
	}
	PG_CATCH();
	{
		if (iterstate)
			ReorderBufferIterTXNFinish(rb, iterstate);
		TeardownHistoricSnapshot(true);
		AbortCurrentTransaction();
		ReorderBufferExecuteInvalidations(rb, txn);
		if (using_subtxn)
			RollbackAndReleaseCurrentSubTransaction();
		if (snapshot_now->copied)
			ReorderBufferFreeSnap(rb, snapshot_now);
		ReorderBufferCleanupTXN(rb, txn);
		PG_RE_THROW();
	}
	PG_END_TRY();
}
void ReorderBufferCommit ( ReorderBuffer * rb , TransactionId xid , XLogRecPtr commit_lsn , XLogRecPtr end_lsn , TimestampTz commit_time , RepOriginId origin_id , XLogRecPtr origin_lsn ) { ReorderBufferTXN * txn ; volatile Snapshot snapshot_now ; volatile CommandId command_id = FirstCommandId ; bool using_subtxn ; ReorderBufferIterTXNState * volatile iterstate = NULL ; txn = ReorderBufferTXNByXid ( rb , xid , false , NULL , InvalidXLogRecPtr , false ) ; if ( txn == NULL ) return ; txn -> final_lsn = commit_lsn ; txn -> end_lsn = end_lsn ; txn -> commit_time = commit_time ; txn -> origin_id = origin_id ; txn -> origin_lsn = origin_lsn ; if ( txn -> base_snapshot == NULL ) { Assert ( txn -> ninvalidations == 0 ) ; ReorderBufferCleanupTXN ( rb , txn ) ; return ; } snapshot_now = txn -> base_snapshot ; ReorderBufferBuildTupleCidHash ( rb , txn ) ; SetupHistoricSnapshot ( snapshot_now , txn -> tuplecid_hash ) ; using_subtxn = IsTransactionOrTransactionBlock ( ) ; PG_TRY ( ) ; { ReorderBufferChange * change ; ReorderBufferChange * specinsert = NULL ; if ( using_subtxn ) BeginInternalSubTransaction ( " " ) ; else StartTransactionCommand ( ) ; rb -> begin ( rb , txn ) ; iterstate = ReorderBufferIterTXNInit ( rb , txn ) ; while ( ( change = ReorderBufferIterTXNNext ( rb , iterstate ) ) != NULL ) { Relation relation = NULL ; Oid reloid ; switch ( change -> action ) { case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM : Assert ( specinsert -> data . tp . oldtuple == NULL ) ; change = specinsert ; change -> action = REORDER_BUFFER_CHANGE_INSERT ; case REORDER_BUFFER_CHANGE_INSERT : case REORDER_BUFFER_CHANGE_UPDATE : case REORDER_BUFFER_CHANGE_DELETE : Assert ( snapshot_now ) ; reloid = RelidByRelfilenode ( change -> data . tp . relnode . spcNode , change -> data . tp . relnode . relNode ) ; if ( reloid == InvalidOid && change -> data . tp . newtuple == NULL && change -> data . tp . oldtuple == NULL ) goto change_done ; else if ( reloid == InvalidOid ) elog ( ERROR , " \" \" " , relpathperm ( change -> data . tp . relnode , MAIN_FORKNUM ) ) ; relation = RelationIdGetRelation ( reloid ) ; if ( relation == NULL ) elog ( ERROR , " \" \" " , reloid , relpathperm ( change -> data . tp . relnode , MAIN_FORKNUM ) ) ; if ( ! RelationIsLogicallyLogged ( relation ) ) goto change_done ; if ( relation -> rd_rel -> relkind == RELKIND_SEQUENCE ) goto change_done ; if ( ! IsToastRelation ( relation ) ) { ReorderBufferToastReplace ( rb , txn , relation , change ) ; rb -> apply_change ( rb , txn , relation , change ) ; if ( change -> data . tp . clear_toast_afterwards ) ReorderBufferToastReset ( rb , txn ) ; } else if ( change -> action == REORDER_BUFFER_CHANGE_INSERT ) { dlist_delete ( & change -> node ) ; ReorderBufferToastAppendChunk ( rb , txn , relation , change ) ; } change_done : if ( specinsert != NULL ) { ReorderBufferReturnChange ( rb , specinsert ) ; specinsert = NULL ; } if ( relation != NULL ) { RelationClose ( relation ) ; relation = NULL ; } break ; case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT : if ( specinsert != NULL ) { ReorderBufferReturnChange ( rb , specinsert ) ; specinsert = NULL ; } dlist_delete ( & change -> node ) ; specinsert = change ; break ; case REORDER_BUFFER_CHANGE_MESSAGE : rb -> message ( rb , txn , change -> lsn , true , change -> data . msg . prefix , change -> data . msg . message_size , change -> data . msg . message ) ; break ; case REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT : TeardownHistoricSnapshot ( false ) ; if ( snapshot_now -> copied ) { ReorderBufferFreeSnap ( rb , snapshot_now ) ; snapshot_now = ReorderBufferCopySnap ( rb , change -> data . snapshot , txn , command_id ) ; } else if ( change -> data . snapshot -> copied ) { snapshot_now = ReorderBufferCopySnap ( rb , change -> data . snapshot , txn , command_id ) ; } else { snapshot_now = change -> data . snapshot ; } SetupHistoricSnapshot ( snapshot_now , txn -> tuplecid_hash ) ; break ; case REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID : Assert ( change -> data . command_id != InvalidCommandId ) ; if ( command_id < change -> data . command_id ) { command_id = change -> data . command_id ; if ( ! snapshot_now -> copied ) { snapshot_now = ReorderBufferCopySnap ( rb , snapshot_now , txn , command_id ) ; } snapshot_now -> curcid = command_id ; TeardownHistoricSnapshot ( false ) ; SetupHistoricSnapshot ( snapshot_now , txn -> tuplecid_hash ) ; ReorderBufferExecuteInvalidations ( rb , txn ) ; } break ; case REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID : elog ( ERROR , " " ) ; break ; } } if ( specinsert ) { ReorderBufferReturnChange ( rb , specinsert ) ; specinsert = NULL ; } ReorderBufferIterTXNFinish ( rb , iterstate ) ; iterstate = NULL ; rb -> commit ( rb , txn , commit_lsn ) ; if ( GetCurrentTransactionIdIfAny ( ) != InvalidTransactionId ) elog ( ERROR , " " , GetCurrentTransactionId ( ) ) ; TeardownHistoricSnapshot ( false ) ; AbortCurrentTransaction ( ) ; ReorderBufferExecuteInvalidations ( rb , txn ) ; if ( using_subtxn ) RollbackAndReleaseCurrentSubTransaction ( ) ; if ( snapshot_now -> copied ) ReorderBufferFreeSnap ( rb , snapshot_now ) ; ReorderBufferCleanupTXN ( rb , txn ) ; } PG_CATCH ( ) ; { if ( iterstate ) ReorderBufferIterTXNFinish ( rb , iterstate ) ; TeardownHistoricSnapshot ( true ) ; AbortCurrentTransaction ( ) ; ReorderBufferExecuteInvalidations ( rb , txn ) ; if ( using_subtxn ) RollbackAndReleaseCurrentSubTransaction ( ) ; if ( snapshot_now -> copied ) ReorderBufferFreeSnap ( rb , snapshot_now ) ; ReorderBufferCleanupTXN ( rb , txn ) ; PG_RE_THROW ( ) ; } PG_END_TRY ( ) ; }
Perform the replay of a transaction and it's non-aborted subtransactions.
Perform the replay of a transaction and it ' s non - aborted subtransactions .
/* unknown transaction, nothing to replay */ /*
	 * If this transaction didn't have any real changes in our database, it's
	 * OK not to have a snapshot. Note that ReorderBufferCommitChild will have
	 * transferred its snapshot to this transaction if it had one and the
	 * toplevel tx didn't.
	 */ /* build data to be able to lookup the CommandIds of catalog tuples */ /* setup the initial snapshot */ /*
	 * Decoding needs access to syscaches et al., which in turn use
	 * heavyweight locks and such. Thus we need to have enough state around to
	 * keep track of those.  The easiest way is to simply use a transaction
	 * internally.  That also allows us to easily enforce that nothing writes
	 * to the database by checking for xid assignments.
	 *
	 * When we're called via the SQL SRF there's already a transaction
	 * started, so start an explicit subtransaction there.
	 */ /*
					 * Confirmation for speculative insertion arrived. Simply
					 * use as a normal record. It'll be cleaned up at the end
					 * of INSERT processing.
					 */ /* intentionally fall through */ /*
					 * Catalog tuple without data, emitted while catalog was
					 * in the process of being rewritten.
					 */ /*
					 * For now ignore sequence changes entirely. Most of the
					 * time they don't log changes using records we
					 * understand, so it doesn't make sense to handle the few
					 * cases we do.
					 */ /* user-triggered change */ /*
						 * Only clear reassembled toast chunks if we're sure
						 * they're not required anymore. The creator of the
						 * tuple tells us.
						 */ /* we're not interested in toast deletions */ /*
						 * Need to reassemble the full toasted Datum in
						 * memory, to ensure the chunks don't get reused till
						 * we're done remove it from the list of this
						 * transaction's changes. Otherwise it will get
						 * freed/reused while restoring spooled data from
						 * disk.
						 */ /*
					 * Either speculative insertion was confirmed, or it was
					 * unsuccessful and the record isn't needed anymore.
					 */ /*
					 * Speculative insertions are dealt with by delaying the
					 * processing of the insert until the confirmation record
					 * arrives. For that we simply unlink the record from the
					 * chain, so it does not get freed/reused while restoring
					 * spooled data from disk.
					 *
					 * This is safe in the face of concurrent catalog changes
					 * because the relevant relation can't be changed between
					 * speculative insertion and confirmation due to
					 * CheckTableNotInUse() and locking.
					 */ /* clear out a pending (and thus failed) speculation */ /* and memorize the pending insertion */ /* get rid of the old */ /*
					 * Restored from disk, need to be careful not to double
					 * free. We could introduce refcounting for that, but for
					 * now this seems infrequent enough not to care.
					 */ /* and continue with the new one */ /* we don't use the global one anymore */ /*
						 * Every time the CommandId is incremented, we could
						 * see new catalog contents, so execute all
						 * invalidations.
						 */ /*
		 * There's a speculative insertion remaining, just clean in up, it
		 * can't have been successful, otherwise we'd gotten a confirmation
		 * record.
		 */ /* clean up the iterator */ /* call commit callback */ /* this is just a sanity check against bad output plugin behaviour */ /* cleanup */ /*
		 * Aborting the current (sub-)transaction as a whole has the right
		 * semantics. We want all locks acquired in here to be released, not
		 * reassigned to the parent and we do not want any database access
		 * have persistent effects.
		 */ /* make sure there's no cache pollution */ /* remove potential on-disk data, and deallocate */ /* TODO: Encapsulate cleanup from the PG_TRY and PG_CATCH blocks */ /*
		 * Force cache invalidation to happen outside of a valid transaction
		 * to prevent catalog access as we just caught an error.
		 */ /* make sure there's no cache pollution */ /* remove potential on-disk data, and deallocate */